lightyear_inputs 0.21.0-rc.3

//! Module to handle client inputs
//!
//! Client inputs are generated by the user and sent to the server.
//! They have to be handled separately from other messages, for several reasons:
//! - the history of inputs might need to be saved on the client to perform rollback and client-prediction
//! - we not only send the input for tick T, but we also include the inputs for the last N ticks before T. This redundancy helps ensure
//!   that the server isn't missing any client inputs even if a packet gets lost
//! - we must provide [`SystemSet`]s so that the user can order their systems before and after the input handling
//!
//! ### Adding a new input type
//!
//! An input type is an enum that implements the `UserAction` trait.
//! This trait is a marker trait that is used to tell Lightyear that this type can be used as an input.
//! In particular inputs must be `Serialize`, `Deserialize`, `Clone` and `PartialEq`.
//!
//! You can then add the input type by adding the `InputPlugin<InputType>` to your app.
//!
//! ```rust
//! use bevy_ecs::entity::{EntityMapper, MapEntities};
//! use serde::{Serialize, Deserialize};
//!
//! #[derive(Serialize, Deserialize, Clone, PartialEq, Debug)]
//! pub enum MyInput {
//!     Move { x: f32, y: f32 },
//!     Jump,
//!     // we need a variant for "no input", to differentiate between "no input" and "missing input packet"
//!     None,
//! }
//!
//! // every input must implement MapEntities
//! impl MapEntities for MyInput {
//!     fn map_entities<M: EntityMapper>(&mut self, entity_mapper: &mut M) {
//!     }
//! }
//! ```
//!
//! ### Sending inputs
//!
//! There are several steps to use the `InputPlugin`:
//! - (optional) read the inputs from an external signal (mouse click or keyboard press, for instance)
//! - to buffer inputs for each tick. This is done by updating the `ActionState` component in a system.
//!   That system must run in the [`InputSet::BufferClientInputs`] system set, in the `FixedPreUpdate` stage.
//! - handle inputs in your game logic in systems that run in the `FixedUpdate` schedule. These systems
//!   will read the inputs using the [`InputBuffer`] component.

use crate::InputChannel;
use crate::config::InputConfig;
use crate::input_buffer::InputBuffer;
use crate::input_message::{ActionStateSequence, InputMessage, InputTarget, PerTargetData};
use crate::plugin::InputPlugin;
#[cfg(feature = "metrics")]
use alloc::format;
use alloc::{vec, vec::Vec};
use bevy_app::{
    App, FixedPostUpdate, FixedPreUpdate, Plugin, PostUpdate, RunFixedMainLoop,
    RunFixedMainLoopSystem,
};
use bevy_ecs::{
    entity::{Entity, MapEntities},
    observer::Trigger,
    query::{Has, With, Without},
    resource::Resource,
    schedule::{IntoScheduleConfigs, SystemSet},
    system::{Commands, Query, Res, ResMut, Single, StaticSystemParam},
};
use lightyear_connection::host::HostClient;
#[cfg(feature = "interpolation")]
use lightyear_core::prelude::Tick;
use lightyear_core::prelude::{NetworkTimeline, Rollback};
use lightyear_core::tick::TickDuration;
#[cfg(feature = "interpolation")]
use lightyear_core::time::TickDelta;
use lightyear_core::timeline::LocalTimeline;
use lightyear_core::timeline::SyncEvent;
#[cfg(feature = "interpolation")]
use lightyear_interpolation::plugin::InterpolationDelay;
#[cfg(feature = "interpolation")]
use lightyear_interpolation::prelude::InterpolationTimeline;
use lightyear_messages::MessageManager;
use lightyear_messages::plugin::MessageSet;
use lightyear_messages::prelude::{MessageReceiver, MessageSender};
use lightyear_prediction::Predicted;
use lightyear_replication::components::{Confirmed, PrePredicted};
use lightyear_sync::plugin::SyncSet;
use lightyear_sync::prelude::InputTimeline;
use lightyear_sync::prelude::client::IsSynced;
use lightyear_transport::channel::ChannelKind;
use lightyear_transport::prelude::ChannelRegistry;
use tracing::{debug, error, trace};

#[derive(SystemSet, Debug, Hash, PartialEq, Eq, Clone, Copy)]
pub enum InputSet {
    // PRE UPDATE
    /// Receive the InputMessage from other clients
    ReceiveInputMessages,

    // FIXED PRE UPDATE
    /// System Set where the user should emit InputEvents, they will be buffered in the InputBuffers in the BufferClientInputs set.
    ///
    /// (For Leafwing, there is nothing to do because the ActionState is updated by leafwing)
    /// (For BEI, there is nothing to do because the `Actions<C>` is updated by the BEI plugin)
    WriteClientInputs,
    /// System Set where we update the ActionState and the InputBuffers
    /// - no rollback: we write the ActionState to the InputBuffers
    /// - rollback: we fetch the ActionState value from the InputBuffers
    BufferClientInputs,

    // FIXED POST UPDATE
    /// Prepare a message for the server with the current tick's inputs.
    /// (we do this in the FixedUpdate schedule because if the simulation is slow (e.g. 10Hz)
    /// we don't want to send an InputMessage every frame)
    PrepareInputMessage,
    /// Restore the ActionState for the correct tick (without InputDelay) from the buffer
    RestoreInputs,

    // POST UPDATE
    /// System Set to prepare the input message
    SendInputMessage,
    /// Clean up old values to prevent the buffers from growing indefinitely
    CleanUp,
}

pub struct ClientInputPlugin<S: ActionStateSequence> {
    config: InputConfig<S::Action>,
}

impl<S: ActionStateSequence> ClientInputPlugin<S> {
    pub fn new(config: InputConfig<S::Action>) -> Self {
        Self { config }
    }
}

impl<S: ActionStateSequence> ClientInputPlugin<S> {
    fn default() -> Self {
        Self::new(InputConfig::default())
    }
}

impl<S: ActionStateSequence + MapEntities> Plugin for ClientInputPlugin<S> {
    fn build(&self, app: &mut App) {
        if !app.is_plugin_added::<InputPlugin<S>>() {
            app.add_plugins(InputPlugin::<S>::default());
        }
        app.insert_resource(self.config);
        app.init_resource::<MessageBuffer<S>>();
        // SETS
        // NOTE: this is subtle! We receive remote players messages after
        //  RunFixedMainLoopSystem::BeforeFixedMainLoop to ensure that leafwing `states` have
        //  been switched to the `fixed_update` state (see https://github.com/Leafwing-Studios/leafwing-input-manager/blob/v0.16/src/plugin.rs#L170)
        //  Conveniently, this also ensures that we run this after InternalMainSet::<ClientMarker>::ReceiveEvents
        app.configure_sets(
            RunFixedMainLoop,
            InputSet::ReceiveInputMessages
                .before(RunFixedMainLoopSystem::FixedMainLoop)
                .after(RunFixedMainLoopSystem::BeforeFixedMainLoop),
        );

        app.configure_sets(
            FixedPreUpdate,
            (InputSet::WriteClientInputs, InputSet::BufferClientInputs).chain(),
        );
        app.configure_sets(
            FixedPostUpdate,
            (
                // we write the InputMessage for the current tick, then restore
                // the correct inputs if there is any input delay
                InputSet::PrepareInputMessage,
                InputSet::RestoreInputs,
            )
                .chain(),
        );
        app.configure_sets(
            PostUpdate,
            ((
                SyncSet::Sync,
                // run after SyncSet to make sure that the TickEvents are handled
                // and that the interpolation_delay injected in the message are correct
                InputSet::SendInputMessage,
                InputSet::CleanUp,
                MessageSet::Send,
            )
                .chain(),),
        );

        // SYSTEMS
        if self.config.rebroadcast_inputs {
            app.add_systems(
                RunFixedMainLoop,
                receive_remote_player_input_messages::<S>.in_set(InputSet::ReceiveInputMessages),
            );
        }
        app.add_systems(
            FixedPreUpdate,
            (buffer_action_state::<S>, get_action_state::<S>)
                .chain()
                .in_set(InputSet::BufferClientInputs),
        );
        app.add_systems(
            FixedPostUpdate,
            (
                // we want:
                // - to write diffs for the delayed tick (in the next FixedUpdate run), so re-fetch the delayed action-state
                //   this is required in case the FixedUpdate schedule runs multiple times in a frame,
                // - next frame's input-map (in PreUpdate) to act on the delayed tick, so re-fetch the delayed action-state
                get_delayed_action_state::<S>.in_set(InputSet::RestoreInputs),
                prepare_input_message::<S>.in_set(InputSet::PrepareInputMessage),
            ),
        );
        app.add_systems(
            PostUpdate,
            (
                clean_buffers::<S>.in_set(InputSet::CleanUp),
                send_input_messages::<S>.in_set(InputSet::SendInputMessage),
            ),
        );
        // if the client tick is updated because of a desync, update the ticks in the input buffers
        app.add_observer(receive_tick_events::<S>);
    }
}

/// Write the value of the ActionState in the InputBuffer.
/// (so that we can pull it for rollback or for delayed inputs)
///
/// If we have input-delay, we will store the current ActionState in the buffer at the delayed-tick,
/// and we will pull ActionStates from the buffer instead of just using the ActionState component directly.
///
/// We do not need to buffer inputs during rollback, as they have already been buffered
fn buffer_action_state<S: ActionStateSequence>(
    context: StaticSystemParam<S::Context>,
    // we buffer inputs even for the Host-Server so that
    // 1. the HostServer client can broadcast inputs to other clients
    // 2. the HostServer client can have input delay
    sender: Single<(&InputTimeline, &LocalTimeline), Without<Rollback>>,
    mut action_state_query: Query<
        (Entity, &S::State, &mut InputBuffer<S::Snapshot>),
        With<S::Marker>,
    >,
) {
    let (input_timeline, local_timeline) = sender.into_inner();
    let current_tick = local_timeline.tick();
    let tick = current_tick + input_timeline.input_delay() as i16;
    for (entity, action_state, mut input_buffer) in action_state_query.iter_mut() {
        InputBuffer::set(
            &mut input_buffer,
            tick,
            S::to_snapshot(action_state, &context),
        );
        trace!(
            ?entity,
            // ?action_state,
            ?current_tick,
            delayed_tick = ?tick,
            input_buffer = %input_buffer.as_ref(),
            "set action state in input buffer",
        );
        #[cfg(feature = "metrics")]
        {
            metrics::gauge!(format!(
                "inputs::{}::{}::buffer_size",
                core::any::type_name::<S::Action>(),
                entity
            ))
            .set(input_buffer.len() as f64);
        }
    }
}

/// Retrieve the ActionState for the current tick.
fn get_action_state<S: ActionStateSequence>(
    context: StaticSystemParam<S::Context>,
    // NOTE: we skip this for host-client because a similar system does the same thing
    //  in the server, and also clears the buffers
    sender: Single<(&LocalTimeline, &InputTimeline, Has<Rollback>), Without<HostClient>>,
    // NOTE: we want to apply the Inputs for BOTH the local player and the remote player.
    // - local player: we need to get the input from the InputBuffer because of input delay
    // - remote player: we want to reduce the amount of rollbacks by updating the ActionState
    //   as fast as possible (the inputs are broadcasted with no delay)
    mut action_state_query: Query<(Entity, &mut S::State, &InputBuffer<S::Snapshot>)>,
) {
    let (local_timeline, input_timeline, is_rollback) = sender.into_inner();
    let input_delay = input_timeline.input_delay() as i16;
    // If there is no rollback and no input_delay, we just buffered the input so there is nothing to do.
    if !is_rollback && input_delay == 0 {
        return;
    }
    let tick = local_timeline.tick();
    for (entity, mut action_state, input_buffer) in action_state_query.iter_mut() {
        // We only apply the ActionState from the buffer if we have one.
        // If we don't (which could happen for remote inputs), we won't do anything.
        // This is equivalent to considering that the remote player will keep playing the last action they played.
        if let Some(snapshot) = input_buffer.get(tick) {
            S::from_snapshot(action_state.as_mut(), snapshot, &context);
            trace!(
                ?entity,
                ?tick,
                // ?action_state,
                "fetched action state from input buffer: {:?}",
                // action_state.get_pressed(),
                input_buffer
            );
        }
    }
}

/// At the start of the frame, restore the ActionState to the latest-action state in buffer
/// (e.g. the delayed action state) because all inputs (i.e. diffs) are applied to the delayed action-state.
fn get_delayed_action_state<S: ActionStateSequence>(
    context: StaticSystemParam<S::Context>,
    sender: Query<(&InputTimeline, &LocalTimeline, Has<Rollback>), With<IsSynced<InputTimeline>>>,
    mut action_state_query: Query<
        (Entity, &mut S::State, &InputBuffer<S::Snapshot>),
        // Filter so that this is only for directly controlled players, not remote players
        With<S::Marker>,
    >,
) {
    let Ok((input_timeline, local_timeline, is_rollback)) = sender.single() else {
        return;
    };
    let input_delay_ticks = input_timeline.input_delay() as i16;
    if is_rollback || input_delay_ticks == 0 {
        return;
    }
    let delayed_tick = local_timeline.tick() + input_delay_ticks;
    for (entity, mut action_state, input_buffer) in action_state_query.iter_mut() {
        // TODO: lots of clone + is complicated. Shouldn't we just have a DelayedActionState component + resource?
        //  the problem is that the Leafwing Plugin works on ActionState directly...
        if let Some(delayed_action_state) = input_buffer.get(delayed_tick) {
            S::from_snapshot(action_state.as_mut(), delayed_action_state, &context);
            trace!(
                ?entity,
                ?delayed_tick,
                // ?action_state,
                "fetched delayed action state from input buffer: {}",
                input_buffer
            );
        }
        // TODO: if we don't find an ActionState in the buffer, should we reset the delayed one to default?
    }
}

/// System that removes old entries from the InputBuffer
fn clean_buffers<S: ActionStateSequence>(
    // NOTE: we skip this for host-client because the get_action_state system on the server
    //  also clears the buffers
    sender: Query<&LocalTimeline, (With<InputTimeline>, Without<HostClient>)>,
    mut input_buffer_query: Query<&mut InputBuffer<S::Snapshot>>,
) {
    let Ok(local_timeline) = sender.single() else {
        return;
    };
    // assuming that we don't rollback more than 20 ticks, or send more than 20 ticks worth of inputs
    let old_tick = local_timeline.tick() - 20;

    // trace!(
    //     "popping all input buffers since old tick: {old_tick:?}",
    // );
    for mut input_buffer in input_buffer_query.iter_mut() {
        input_buffer.pop(old_tick);
    }
}

// TODO: is this actually necessary? The sync happens in PostUpdate,
//  so maybe it's ok if the InputMessages contain the pre-sync tick! (since those inputs happened
//  before the sync). If it's not needed, send the messages directly in FixedPostUpdate!
//  Actually maybe it is, because the send-tick on the server will be updated.
/// Buffer that will store the InputMessages we want to write this frame.
///
/// We need this because:
/// - we write the InputMessages during FixedPostUpdate
/// - we apply the TickUpdateEvents (from doing sync) during PostUpdate, which might affect the ticks from the InputMessages.
///   During this phase, we want to update the tick of the InputMessages that we wrote during FixedPostUpdate.
#[derive(Debug, Resource)]
pub(crate) struct MessageBuffer<S>(Vec<InputMessage<S>>);

impl<A> Default for MessageBuffer<A> {
    fn default() -> Self {
        Self(vec![])
    }
}

/// Take the input buffer, and prepare the input message to send to the server
fn prepare_input_message<S: ActionStateSequence>(
    mut message_buffer: ResMut<MessageBuffer<S>>,
    tick_duration: Res<TickDuration>,
    input_config: Res<InputConfig<S::Action>>,
    sender: Single<
        (
            &LocalTimeline,
            &InputTimeline,
            &MessageManager,
            Has<HostClient>,
        ),
        // the host-client doesn't need to send input messages since the ActionState is already on the entity
        // unless we want to rebroadcast the HostClient inputs to other clients (in which
        // case we prepare the input-message, which will be send_local to the server)
        (With<IsSynced<InputTimeline>>, Without<Rollback>),
    >,
    channel_registry: Res<ChannelRegistry>,
    input_buffer_query: Query<
        (
            Entity,
            &InputBuffer<S::Snapshot>,
            Option<&Predicted>,
            Option<&PrePredicted>,
        ),
        With<S::Marker>,
    >,
) {
    let (local_timeline, input_timeline, message_manager, is_host_client) = sender.into_inner();
    if is_host_client && !input_config.rebroadcast_inputs {
        // the host-client doesn't need to send input messages since the ActionState is already on the entity
        // unless we want to rebroadcast the HostClient inputs to other clients
        return;
    }

    // we send a message from the latest tick that we have available, which is the delayed tick
    let current_tick = local_timeline.tick();
    let tick = current_tick + input_timeline.input_delay() as i16;
    // TODO: the number of messages should be in SharedConfig
    trace!(delayed_tick = ?tick, ?current_tick, "prepare_input_message");
    // TODO: instead of redundancy, send ticks up to the latest yet ACK-ed input tick
    //  this means we would also want to track packet->message acks for unreliable channels as well, so we can notify
    //  this system what the latest acked input tick is?

    let input_send_interval = channel_registry
        .settings(ChannelKind::of::<InputChannel>())
        .unwrap()
        .send_frequency;
    // we send redundant inputs, so that if a packet is lost, we can still recover
    // A redundancy of 2 means that we can recover from 1 lost packet
    let mut num_tick: u16 = ((input_send_interval.as_nanos() / tick_duration.as_nanos()) + 1)
        .try_into()
        .unwrap();
    num_tick *= input_config.packet_redundancy;
    let mut message = InputMessage::<S>::new(tick);
    for (entity, input_buffer, predicted, pre_predicted) in input_buffer_query.iter() {
        trace!(
            ?tick,
            ?entity,
            "Preparing input message with buffer: {:?}",
            input_buffer
        );

        // Make sure that server can read the inputs correctly
        // TODO: currently we are not sending inputs for pre-predicted entities until we receive the confirmation from the server
        //  could we find a way to do it?
        //  maybe if it's pre-predicted, we send the original entity (pre-predicted), and the server will apply the conversion
        //   on their end?
        if let Some(target) = if is_host_client {
            // we are using PrePredictedEntity to make sure that MapEntities will be used on the client receiving side
            Some(InputTarget::PrePredictedEntity(entity))
        } else if pre_predicted.is_some() {
            // wait until the client receives the PrePredicted entity confirmation to send inputs
            // otherwise we get failed entity_map logs
            // TODO: the problem is that we wait until we have received the server answer. Ideally we would like
            //  to wait until the server has received the PrePredicted entity
            if predicted.is_none() {
                continue;
            }
            trace!(
                ?tick,
                "sending inputs for pre-predicted entity! Local client entity: {:?}", entity
            );
            // TODO: not sure if this whole pre-predicted inputs thing is worth it, because the server won't be able to
            //  to receive the inputs until it receives the pre-predicted spawn message.
            //  so all the inputs sent between pre-predicted spawn and server-receives-pre-predicted will be lost

            // TODO: I feel like pre-predicted inputs work well only for global-inputs, because then the server can know
            //  for which client the inputs were!

            // 0. the entity is pre-predicted, no need to convert the entity (the mapping will be done on the server, when
            // receiving the message. It's possible because the server received the PrePredicted entity before)
            Some(InputTarget::PrePredictedEntity(entity))
        } else {
            // 1. if the entity is confirmed, we need to convert the entity to the server's entity
            // 2. if the entity is predicted, we need to first convert the entity to confirmed, and then from confirmed to remote
            if let Some(confirmed) = predicted.map_or(Some(entity), |p| p.confirmed_entity) {
                message_manager.entity_mapper.get_remote(confirmed).map(|server_entity|{
                    trace!("sending input for server entity: {:?}. local entity: {:?}, confirmed: {:?}", server_entity, entity, confirmed);
                    InputTarget::Entity(server_entity)
                })
            } else {
                // TODO: entity is not predicted or not confirmed? also need to do the conversion, no?
                trace!("not sending inputs because couldnt find server entity");
                None
            }
        } {
            if let Some(state_sequence) = S::build_from_input_buffer(input_buffer, num_tick, tick) {
                message.inputs.push(PerTargetData {
                    target,
                    states: state_sequence,
                });
            }
        }
    }

    // we send a message even when there are 0 inputs because that itself is information
    trace!(
        ?tick,
        ?num_tick,
        "sending input message for {:?}: {:?}",
        core::any::type_name::<S::Action>(),
        message
    );
    message_buffer.0.push(message);

    // NOTE: keep the older input values in the InputBuffer! because they might be needed when we rollback for client prediction
}

/// Read the InputMessages of other clients from the server to update their InputBuffer and ActionState.
/// This is useful if we want to do client-prediction for remote players.
///
/// If the InputBuffer/ActionState is missing, we will add it.
///
/// We will apply the diffs on the Predicted entity.
fn receive_remote_player_input_messages<S: ActionStateSequence>(
    mut commands: Commands,
    link: Single<
        (
            &MessageManager,
            &mut MessageReceiver<InputMessage<S>>,
            &LocalTimeline,
        ),
        With<IsSynced<InputTimeline>>,
    >,
    // TODO: currently we do not handle entities that are controlled by multiple clients
    confirmed_query: Query<&Confirmed, Without<S::Marker>>,
    mut predicted_query: Query<
        Option<&mut InputBuffer<S::Snapshot>>,
        // the host-client won't receive input messages from the Server
        (With<Predicted>, Without<S::Marker>, Without<HostClient>),
    >,
) {
    let (manager, mut receiver, timeline) = link.into_inner();
    let tick = timeline.tick();
    receiver.receive().for_each(|message| {
        trace!(?message.end_tick, ?message, "received remote input message for action: {:?}", core::any::type_name::<S::Action>());
        for target_data in message.inputs {
            // - the input target has already been set to the server entity in the InputMessage
            // - it has been mapped to a client-entity on the client during deserialization
            //   ONLY if it's PrePredicted (look at the MapEntities implementation)
            let entity = match target_data.target {
                InputTarget::Entity(entity) => {
                    // TODO: find a better way!
                    // if InputTarget = Entity, we still need to do the mapping
                    manager
                        .entity_mapper
                        .get_local(entity)
                }
                InputTarget::PrePredictedEntity(entity) => Some(entity),
            };
            if let Some(entity) = entity {
                debug!(
                    "received input message for entity: {:?}. Applying to diff buffer.",
                    entity
                );
                if let Ok(confirmed) = confirmed_query.get(entity) {
                    if let Some(predicted) = confirmed.predicted {
                        if let Ok(input_buffer) = predicted_query.get_mut(predicted) {
                            trace!(confirmed= ?entity, ?predicted, end_tick = ?message.end_tick, "update action diff buffer for remote player PREDICTED using input message");
                            if let Some(mut input_buffer) = input_buffer {
                                target_data.states.update_buffer(&mut input_buffer, message.end_tick);
                                #[cfg(feature = "metrics")]
                                {
                                    let margin = input_buffer.end_tick().unwrap() - tick;
                                    metrics::gauge!(format!(
                                                    "inputs::{}::remote_player::{}::buffer_margin",
                                                    core::any::type_name::<S::Action>(),
                                                    entity
                                                ))
                                        .set(margin as f64);
                                    metrics::gauge!(format!(
                                                    "inputs::{}::remote_player::{}::buffer_size",
                                                    core::any::type_name::<S::Action>(),
                                                    entity
                                                ))
                                        .set(input_buffer.len() as f64);
                                }
                            } else {
                                // add the ActionState or InputBuffer if they are missing
                                let mut input_buffer = InputBuffer::<S::Snapshot>::default();
                                target_data.states.update_buffer(&mut input_buffer, message.end_tick);
                                // if the remote_player's predicted entity doesn't have the InputBuffer, we need to insert them
                                commands.entity(predicted).insert((
                                    input_buffer,
                                    S::State::default()
                                ));
                            };
                        }
                    }
                } else {
                    error!(?entity, ?target_data.states, end_tick = ?message.end_tick, "received input message for unrecognized entity");
                }
            } else {
                error!("received remote player input message for unrecognized entity");
            }
        }
    });
}

/// Drain the messages from the buffer and send them to the server
fn send_input_messages<S: ActionStateSequence>(
    input_config: Res<InputConfig<S::Action>>,
    mut message_buffer: ResMut<MessageBuffer<S>>,
    sender: Single<
        (&mut MessageSender<InputMessage<S>>, Has<HostClient>),
        With<IsSynced<InputTimeline>>,
    >,
    #[cfg(feature = "interpolation")] interpolation_query: Single<
        (&InputTimeline, &InterpolationTimeline),
        (
            With<IsSynced<InterpolationTimeline>>,
            With<IsSynced<InputTimeline>>,
        ),
    >,
) {
    let (mut sender, is_host_client) = sender.into_inner();
    if is_host_client && !input_config.rebroadcast_inputs {
        // the host-client doesn't need to send input messages since the ActionState is already on the entity
        // unless we want to rebroadcast the HostClient inputs to other clients
        message_buffer.0.clear();
        return;
    }
    trace!(
        "Number of input messages to send: {:?}",
        message_buffer.0.len()
    );

    #[cfg(feature = "interpolation")]
    let interpolation_delay = {
        let (input_timeline, interpolation_timeline) = interpolation_query.into_inner();

        // NOTE: this can be negative because of input-delay!
        let mut delay = input_timeline.now() - interpolation_timeline.now();
        if delay.is_negative() {
            delay = TickDelta::from(Tick(0));
        }
        InterpolationDelay {
            delay: delay.into(),
        }
    };

    #[cfg_attr(
        not(feature = "interpolation"),
        expect(unused_mut, reason = "Used by expression behind feature flag")
    )]
    for mut message in message_buffer.0.drain(..) {
        // if lag compensation is enabled, we send the current delay to the server
        // (this runs here because the delay is only correct after the SyncSet has run)
        // TODO: or should we actually use the interpolation_delay BEFORE SyncSet
        //  because the user is reacting to stuff from the previous frame?
        #[cfg(feature = "interpolation")]
        if input_config.lag_compensation {
            message.interpolation_delay = Some(interpolation_delay);
        }
        sender.send::<InputChannel>(message);
    }
}

/// In case the client tick changes suddenly, we also update the InputBuffer accordingly
fn receive_tick_events<S: ActionStateSequence>(
    trigger: Trigger<SyncEvent<InputTimeline>>,
    mut message_buffer: ResMut<MessageBuffer<S>>,
    mut input_buffer_query: Query<&mut InputBuffer<S::Snapshot>>,
) {
    let delta = trigger.tick_delta;
    for mut input_buffer in input_buffer_query.iter_mut() {
        if let Some(start_tick) = input_buffer.start_tick {
            input_buffer.start_tick = Some(start_tick + delta);
            debug!(
                "Receive tick snap event {:?}. Updating input buffer start_tick to {:?}!",
                trigger.event(),
                input_buffer.start_tick
            );
        }
    }
    for message in message_buffer.0.iter_mut() {
        message.end_tick = message.end_tick + delta;
    }
}