lightyear_inputs 0.29.0

//! 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 [`InputSystems::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::config::{InputConfig, SharedInputConfig};
use crate::input_buffer::InputBuffer;
use crate::input_message::{
    ActionStateQueryData, ActionStateSequence, InputMessage, InputSnapshot, InputTarget,
    PerTargetData, StateMut, StateRef,
};
#[cfg(feature = "metrics")]
use crate::metric_handles::InputMetricHandles;
use crate::plugin::InputPlugin;
use crate::{HISTORY_DEPTH, InputChannel};
use alloc::{vec, vec::Vec};
use bevy_app::{
    App, FixedPostUpdate, FixedPreUpdate, Plugin, PostUpdate, PreUpdate, RunFixedMainLoopSystems,
};
use bevy_ecs::entity::{MapEntities, UniqueEntitySlice};
use bevy_ecs::prelude::*;
#[cfg(feature = "interpolation")]
use bevy_time::Fixed;
use bevy_time::{Real, Time, Timer, TimerMode};
use bevy_utils::prelude::DebugName;
use lightyear_connection::network_topology::{NetworkTopology, NetworkingMetadata};
use lightyear_core::prelude::*;
use lightyear_core::tick::TickDuration;
#[cfg(feature = "interpolation")]
use lightyear_core::time::TickDelta;

#[cfg(feature = "interpolation")]
use lightyear_interpolation::prelude::*;
use lightyear_messages::multi::MultiMessageSender;
use lightyear_messages::plugin::MessageSystems;
#[cfg(feature = "prediction")]
use lightyear_messages::prelude::MessageReceiver;
use lightyear_messages::prelude::MessageSender;
use lightyear_prediction::prelude::*;
use lightyear_replication::prelude::{ControlledBy, PreSpawned};
use lightyear_sync::plugin::SyncSystems;
use lightyear_sync::prelude::{InputTimelineConfig, LocalTimelineSync, SyncedLocalTimeline};
use lightyear_transport::prelude::ChannelRegistry;
#[allow(unused_imports)]
use tracing::{debug, error, info, trace, warn};

#[deprecated(note = "Use InputSystems instead")]
pub type InputSet = InputSystems;
#[derive(SystemSet, Debug, Hash, PartialEq, Eq, Clone, Copy)]
pub enum InputSystems {
    // RUN-FIXED-MAIN-LOOP 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,
    // TODO: could this run in RunFixedMainLoop::AfterFixedMainLoop?
    /// Restore the ActionState for the correct tick (without InputDelay) from the buffer
    RestoreInputs,

    // POST UPDATE
    /// Update the metadata where we store information about remote inputs that we received, for instance
    /// the [`LastConfirmedInput`] across all remote clients
    UpdateRemoteInputTicks,
    /// System Set to prepare the input message
    SendInputMessage,
    /// Clean up old values to prevent the buffers from growing indefinitely
    CleanUp,
}

/// Client-side plugin that buffers local action state each tick and sends
/// compressed input messages to the server.
///
/// The plugin also handles:
/// - **Input delay**: buffering inputs ahead of time so the server receives
///   them before it needs them.
/// - **Redundancy**: each message includes the last N ticks of input to
///   recover from packet loss.
/// - **Remote player prediction**: processes other players' inputs when a conventional server
///   rebroadcasts them, or directly from every peer in a P2P topology.
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.init_resource::<SharedInputConfig>();
        app.insert_resource(self.config);
        app.init_resource::<MessageBuffer<S>>();
        // Client input systems may be installed in a combined client/server app. Keep their
        // global clock parameters available even when no client link is active; the
        // SyncedLocalTimeline parameter still gates systems that require synchronization.
        app.init_resource::<LocalTimelineSync>();
        app.init_resource::<InputTimelineConfig>();
        #[cfg(feature = "prediction")]
        {
            if !app.is_plugin_added::<LastConfirmedInputPlugin>() {
                app.add_plugins(LastConfirmedInputPlugin);
            }
        }

        // SETS

        // NOTE: this is subtle! We receive remote players messages after
        //  RunFixedMainLoopSystems::BeforeFixedMainLoop to ensure that the local 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)
        //  We can move this system back in PreUpdate if we drop leafwing support.
        //  Conveniently, this also ensures that we run this after MessageSet::Receive.
        app.configure_sets(
            PreUpdate,
            InputSystems::ReceiveInputMessages
                .before(RunFixedMainLoopSystems::FixedMainLoop)
                .after(RunFixedMainLoopSystems::BeforeFixedMainLoop),
        );

        app.configure_sets(
            FixedPreUpdate,
            (
                InputSystems::WriteClientInputs,
                InputSystems::BufferClientInputs,
            )
                .chain(),
        );
        app.configure_sets(FixedPostUpdate, InputSystems::RestoreInputs);
        app.configure_sets(
            PostUpdate,
            ((
                InputSystems::PrepareInputMessage,
                SyncSystems::Sync,
                // Run after synchronization so local timeline shifts have been handled.
                // and that the interpolation_delay injected in the message are correct
                InputSystems::SendInputMessage,
                InputSystems::CleanUp,
                MessageSystems::Send,
            )
                .chain(),),
        );

        // SYSTEMS
        #[cfg(feature = "prediction")]
        {
            // The topology is a runtime choice. Conventional clients still return immediately
            // when rebroadcasting is disabled, while P2P peers must always drain direct remote
            // input messages even though their client/server rebroadcast option is false.
            // NOTE: we do NOT need to run this after RunFixedMainLoopSystems::BeforeFixedMainLoop to ensure that the
            //  local 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)
            //  because when applying the diffs we manually update the fixed_update_state .

            //  because when we apply the diffs, we start by taking the initial `start_state` snapshot from the message.
            //  That snapshot has both `state` = `fixed_update_state` since it was buffered by the client
            //  during FixedUpdate.
            //  Then, the diffs will update `state`, and that value will be stored in the buffer.
            //
            //  The ActionState will only be modified during FixedPreUpdate in `get_action_state`, so we will have
            //  `state` = `fixed_update_state` in the ActionState component.
            app.configure_sets(
                PreUpdate,
                InputSystems::ReceiveInputMessages
                    .after(MessageSystems::Receive)
                    // NOTE: there is no point in running this after ReplicationSet::Receive, because even if we spawned the entity
                    //  before the corresponding input entity, entity-mapping is applied in MessageSet::Receive and would fail
                    .before(RollbackSystems::Check),
            );
            app.add_systems(
                PreUpdate,
                receive_remote_player_input_messages::<S>
                    .in_set(InputSystems::ReceiveInputMessages),
            );
            app.configure_sets(
                PostUpdate,
                InputSystems::UpdateRemoteInputTicks.after(SyncSystems::Sync),
            );
            app.add_systems(
                PostUpdate,
                update_last_confirmed_input::<S>.in_set(InputSystems::UpdateRemoteInputTicks),
            );
        }
        app.add_systems(
            FixedPreUpdate,
            (buffer_action_state::<S>, get_action_state::<S>)
                .chain()
                .in_set(InputSystems::BufferClientInputs),
        );
        app.add_systems(
            // This should be fine because the user should use the ActionState value in FixedUpdate
            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(InputSystems::RestoreInputs),
            ),
        );
        app.add_systems(
            PostUpdate,
            (
                // TODO: instead, store directly in MessageSender, SyncSet::Sync before MessageSet::Send and register an observer to update the ticks from MessageSender directly!
                prepare_input_message::<S>.in_set(InputSystems::PrepareInputMessage),
                clean_buffers::<S>.in_set(InputSystems::CleanUp),
                send_input_messages::<S>.in_set(InputSystems::SendInputMessage),
            ),
        );
        // if the client tick is updated because of a desync, update the ticks in the input buffers
        app.add_observer(receive_local_timeline_shift::<S>);
    }
}

/// Installs application-global confirmed-input frontier bookkeeping.
///
/// A client application can register several input types. Installing this non-generic plugin once
/// ensures their contributions are finalized only after every input type has updated the shared
/// [`LastConfirmedInput`] resource.
#[cfg(feature = "prediction")]
struct LastConfirmedInputPlugin;

#[cfg(feature = "prediction")]
impl Plugin for LastConfirmedInputPlugin {
    fn build(&self, app: &mut App) {
        app.init_resource::<LastConfirmedInput>();
        app.add_systems(
            PostUpdate,
            finalize_last_confirmed_input.after(InputSystems::UpdateRemoteInputTicks),
        );
    }
}

#[cfg(feature = "prediction")]
/// Preserve the completed aggregate for next frame's rollback decision.
fn finalize_last_confirmed_input(mut last_confirmed_input: ResMut<LastConfirmedInput>) {
    last_confirmed_input.finalize_frame();
}

/// Cached input routing derived from [`NetworkingMetadata`].
///
/// Client/server and host-client modes retain their single Link as the local ownership
/// identity. P2P mode borrows the already-cached connected Link slice and treats `S::Marker` as
/// the local-input identity because each Link represents a remote peer.
#[derive(Clone, Copy, Debug)]
enum InputRoute<'a> {
    ClientServer { link: Entity, host_client: bool },
    P2P(&'a [Entity]),
}

impl<'a> InputRoute<'a> {
    #[inline]
    fn from_topology(topology: &'a NetworkTopology) -> Option<Self> {
        match topology {
            NetworkTopology::Client(link) => Some(Self::ClientServer {
                link: *link,
                host_client: false,
            }),
            NetworkTopology::HostClient { client, .. } => Some(Self::ClientServer {
                link: *client,
                host_client: true,
            }),
            NetworkTopology::P2P { connected, .. } if !connected.is_empty() => {
                Some(Self::P2P(connected.as_slice()))
            }
            NetworkTopology::Undefined
            | NetworkTopology::Server(_)
            | NetworkTopology::P2P { .. }
            | NetworkTopology::Invalid(_) => None,
        }
    }

    #[inline]
    fn is_host_client(self) -> bool {
        matches!(
            self,
            Self::ClientServer {
                host_client: true,
                ..
            }
        )
    }

    #[inline]
    fn requires_prespawned_targets(self) -> bool {
        matches!(self, Self::P2P(_))
    }

    #[inline]
    fn accepts_local_target(self, controlled_by: Option<&ControlledBy>) -> bool {
        match self {
            Self::ClientServer { link, .. } => {
                !controlled_by.is_some_and(|controlled_by| controlled_by.owner != link)
            }
            // `S::Marker` is the local-capture marker in P2P. `ControlledBy` describes a
            // replication sender and therefore cannot identify this app among remote peer Links.
            Self::P2P(_) => true,
        }
    }
}

/// Select the wire identity for one locally controlled input entity.
///
/// Conventional client/server links can use their replication-populated entity map. P2P has no
/// authoritative replication stream to populate those per-Link maps, so every target must use a
/// stable [`PreSpawned`] hash that the other peers can resolve in their own worlds.
fn input_target(
    route: InputRoute<'_>,
    entity: Entity,
    pre_spawned: Option<&PreSpawned>,
) -> InputTarget {
    if let Some(hash) = pre_spawned.and_then(|pre_spawned| pre_spawned.hash) {
        debug!(?hash, ?entity, "Sending input for prespawned entity");
        return InputTarget::PreSpawned(hash);
    }
    assert!(
        !route.requires_prespawned_targets(),
        "P2P input target {entity:?} must have a PreSpawned component with a resolved hash"
    );
    InputTarget::Entity(entity)
}

#[inline]
fn matches_prespawned_target(
    hash: u64,
    p2p_link: Option<Entity>,
    pre_spawned: &PreSpawned,
) -> bool {
    pre_spawned.hash.is_some_and(|candidate| candidate == hash)
        && p2p_link.is_none_or(|link| pre_spawned.receiver == Some(link))
}

// equivalent to &ActionState<S::Action>

/// 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!
///
/// We only buffer inputs after the timelines have synced, otherwise it would be possible to buffer
/// inputs that would not arrive on time in the server (i.e. the server would simulate tick T
/// before receiving the input for tick T).
fn buffer_action_state<S: ActionStateSequence>(
    #[cfg(feature = "metrics")] metric_handles: Res<InputMetricHandles<S>>,
    // 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
    timeline: SyncedLocalTimeline,
    metadata: Res<NetworkingMetadata>,
    rollback: Option<Res<Rollback>>,
    mut action_state_query: Query<
        (
            Entity,
            StateRef<S>,
            &mut InputBuffer<S::Snapshot, S::Action>,
            Option<&ControlledBy>,
        ),
        (With<S::Marker>, Allow<PredictionDisable>),
    >,
) {
    let Some(route) = InputRoute::from_topology(&metadata.mode) else {
        return;
    };
    if rollback.is_some() {
        return;
    }
    let current_tick = timeline.current_tick();
    let tick = current_tick + timeline.input_delay() as i32;
    for (entity, action_state, mut input_buffer, controlled_by) in action_state_query.iter_mut() {
        if !route.accepts_local_target(controlled_by) {
            continue;
        }
        input_buffer.set(tick, S::to_snapshot(action_state));
        trace!(
            ?entity,
            // ?action_state,
            ?current_tick,
            delayed_tick = ?tick,
            input_buffer = %input_buffer.as_ref(),
            "set action state in input buffer",
        );
        trace!(
            target: "lightyear_debug::input",
            kind = "buffer_action_state",
            schedule = "FixedPreUpdate",
            sample_point = "FixedPreUpdate",
            entity = ?entity,
            action = ?DebugName::type_name::<S::Action>(),
            local_tick = current_tick.0,
            input_tick = tick.0,
            buffer_len = input_buffer.len(),
            input_buffer = %input_buffer.as_ref(),
            "buffered local action state"
        );
        #[cfg(feature = "metrics")]
        metric_handles
            .buffer_size(entity)
            .set(input_buffer.len() as f64);
    }
}

/// Retrieve the ActionState for the current tick.
fn get_action_state<S: ActionStateSequence>(
    tick_duration: Res<TickDuration>,
    config: Res<InputConfig<S::Action>>,
    local_timeline: Res<LocalTimeline>,
    // NOTE: we skip this for host-client because a similar system does the same thing
    //  in the server, and also clears the buffers
    input_timeline: Res<LocalTimelineSync>,
    input_timeline_config: Res<InputTimelineConfig>,
    metadata: Res<NetworkingMetadata>,
    rollback: Option<Res<Rollback>>,

    // NOTE: we want to apply the Inputs for BOTH the local player and remote player
    // - local player: we need to get the input from the InputBuffer because of input delay
    // - remote player: during rollbacks, we need to fetch the ActionState from the InputBuffer
    // (for the remote players, we update the ActionState as soon as we receive the RemoteMessage.)
    //  TODO: We could maybe have some decay logic where the input decays to the middle)
    mut action_state_query: Query<
        (
            Entity,
            StateMut<S>,
            &mut InputBuffer<S::Snapshot, S::Action>,
            Has<S::Marker>,
        ),
        Allow<PredictionDisable>,
    >,
) {
    let Some(route) = InputRoute::from_topology(&metadata.mode) else {
        return;
    };
    // A host-client shares its authoritative input application with the server-side input
    // pipeline, so the client system must retain its existing exclusion.
    if route.is_host_client() {
        return;
    }
    let is_rollback = rollback.is_some();
    let input_delay = input_timeline.input_delay() as i32;
    let tick = local_timeline.tick();
    if is_rollback && config.ignore_rollbacks {
        return;
    }
    // TODO!: if config.rebroadcast = False, we don't need to handle remote players, try to encode that statically!
    for (entity, action_state, mut input_buffer, is_local) in action_state_query.iter_mut() {
        if !is_rollback && is_local && input_delay == 0 {
            // for local clients, if there is no rollback and no input_delay:
            // we just buffered the input for the current tick so the action state is already up to date
            continue;
        }

        // NOTE: for remote players:
        // - if we receive a remote input from an older tick (because of prediction), then upon receipt we update the ActionState
        //   immediately so that we can trigger a rollback (in `receive_input_message`) and the ActionState is correctly set
        //   to the past state for the rollback
        // - we cannot just rely on that, because in some cases (lockstep), we receive the remote input in the future, so we need
        //   to read from the buffer to restore the ActionState to the correct tick.
        // - we cannot use `is_lockstep` to check if we receive inputs in the future, because there are cases in non-lockstep where
        //   we could receive some inputs from the future, if we had a high enough input delay.

        // NOTE: we use `get` and not `get_predict ` here.
        // This means that we try to get the value for this exact tick.
        // - For local inputs with input_delay: our current state is in the future, so we need to fetch the exact value from the buffer
        // - For remote inputs with lockstep: the last input is in the future, so we need to fetch the exact value from the buffer
        // - For remote inputs without lockstep: we might receive a message that updates our input buffer, and the last input is
        //   in the past. We already updated the ActionState in `receive_input_message` for that past tick, which means we are
        //   predicting that the action hasn't changed since. We just need to decay it (for rollback or without rollback)
        if let Some(snapshot) = input_buffer.get(tick) {
            // TODO: should we decay_tick the snapshot?
            S::from_snapshot(S::State::into_inner(action_state), snapshot);
            trace!(
                ?entity,
                ?tick,
                ?is_local,
                ?snapshot,
                // ?action_state,
                "fetched action state from input buffer: {:?}",
                // action_state.get_pressed(),
                input_buffer
            );
            trace!(
                target: "lightyear_debug::input",
                kind = "get_action_state",
                schedule = "FixedPreUpdate",
                sample_point = "FixedPreUpdate",
                entity = ?entity,
                action = ?DebugName::type_name::<S::Action>(),
                local_tick = tick.0,
                input_tick = tick.0,
                is_local,
                is_rollback,
                snapshot = ?snapshot,
                buffer_len = input_buffer.len(),
                input_buffer = %*input_buffer,
                "restored action state from input buffer"
            );
        } else if !is_local && (config.rebroadcast_inputs || matches!(route, InputRoute::P2P(_))) {
            if input_timeline_config.is_lockstep() {
                error!("We are in lockstep mode but didn't receive an input for tick {tick:?}!");
            }
            // we are here if:
            // - we are in rollback and we reach a tick further than the last tick we received from the remote
            // - we are not in rollback, in which case we want to decay the ActionState
            let mut snapshot = S::to_snapshot(S::State::as_read_only(&action_state));
            snapshot.decay_tick(tick_duration.0);
            trace!(
                ?entity,
                ?tick,
                "Action = {}, For remote input; no input for tick so we decay the ActionState to: {:?}",
                DebugName::type_name::<S::Action>(),
                snapshot
            );
            trace!(
                target: "lightyear_debug::input",
                kind = "decay_missing_remote_action_state",
                schedule = "FixedPreUpdate",
                sample_point = "FixedPreUpdate",
                entity = ?entity,
                action = ?DebugName::type_name::<S::Action>(),
                local_tick = tick.0,
                input_tick = tick.0,
                is_rollback,
                snapshot = ?snapshot,
                buffer_len = input_buffer.len(),
                "decayed missing remote action state"
            );
            // update the action state with decay
            S::from_snapshot(S::State::into_inner(action_state), &snapshot);
            // add the new snapshot in the buffer
            input_buffer.set(tick, snapshot);
        }
    }
}

/// 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>(
    timeline: SyncedLocalTimeline,
    metadata: Res<NetworkingMetadata>,
    rollback: Option<Res<Rollback>>,
    mut action_state_query: Query<
        (
            Entity,
            StateMut<S>,
            &InputBuffer<S::Snapshot, S::Action>,
            Option<&ControlledBy>,
        ),
        // Filter so that this is only for directly controlled players, not remote players
        (With<S::Marker>, Allow<PredictionDisable>),
    >,
) {
    let Some(route) = InputRoute::from_topology(&metadata.mode) else {
        return;
    };
    let is_rollback = rollback.is_some();
    let input_delay_ticks = timeline.input_delay() as i32;
    if is_rollback || input_delay_ticks == 0 {
        return;
    }
    let tick = timeline.tick();
    let delayed_tick = tick + input_delay_ticks;
    for (entity, action_state, input_buffer, controlled_by) in action_state_query.iter_mut() {
        if !route.accepts_local_target(controlled_by) {
            continue;
        }
        // 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(S::State::into_inner(action_state), delayed_action_state);
            trace!(
                ?entity,
                ?delayed_tick,
                // ?action_state,
                "fetched delayed action state from input buffer: {}",
                input_buffer
            );
            trace!(
                target: "lightyear_debug::input",
                kind = "get_delayed_action_state",
                schedule = "RunFixedMainLoop",
                sample_point = "RunFixedMainLoop",
                entity = ?entity,
                action = ?DebugName::type_name::<S::Action>(),
                local_tick = tick.0,
                input_tick = delayed_tick.0,
                snapshot = ?delayed_action_state,
                buffer_len = input_buffer.len(),
                input_buffer = %input_buffer,
                "restored delayed action state"
            );
        }
        // 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>(
    timeline: Res<LocalTimeline>,
    // NOTE: we skip this for host-client because the get_action_state system on the server
    //  also clears the buffers
    metadata: Res<NetworkingMetadata>,
    prediction_manager: Option<Res<PredictionManager>>,
    input_config: Res<InputTimelineConfig>,
    mut input_buffer_query: Query<
        &mut InputBuffer<S::Snapshot, S::Action>,
        Allow<PredictionDisable>,
    >,
) {
    let Some(route) = InputRoute::from_topology(&metadata.mode) else {
        return;
    };
    if route.is_host_client() {
        return;
    }
    let old_tick = timeline.tick()
        - input_history_depth(
            prediction_manager
                .as_deref()
                .map(|manager| (manager, input_config.as_ref())),
        );

    // 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);
        //  for now do NOT spawn Transform, instead directly use Position/Rotation!
    }
}

fn input_history_depth(
    prediction_manager: Option<(&PredictionManager, &InputTimelineConfig)>,
) -> u32 {
    prediction_manager
        .map(|(manager, input_config)| {
            u32::from(
                manager
                    .rollback_policy
                    .effective_max_rollback_ticks(input_config),
            ) + 1
        })
        .unwrap_or(0)
        .max(HISTORY_DEPTH)
}

// 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.
///
/// This runs once per frame in PostUpdate. It needs to run before SyncSet::Sync, because we buffer
/// the input in a MessageBuffer, and if a `LocalTimelineShift` triggers, we want to adjust the
/// ticks from the `InputMessage`s.
fn prepare_input_message<S: ActionStateSequence>(
    mut message_buffer: ResMut<MessageBuffer<S>>,
    tick_duration: Res<TickDuration>,
    input_config: Res<InputConfig<S::Action>>,
    timeline: SyncedLocalTimeline,
    metadata: Res<NetworkingMetadata>,
    rollback: Option<Res<Rollback>>,
    _channel_registry: Res<ChannelRegistry>,
    input_buffer_query: Query<
        (
            Entity,
            &InputBuffer<S::Snapshot, S::Action>,
            Option<&PreSpawned>,
            Option<&ControlledBy>,
        ),
        (With<S::Marker>, Allow<PredictionDisable>),
    >,
    real_time: Res<Time<Real>>,
    mut send_timer: Local<Option<Timer>>,
) {
    let Some(route) = InputRoute::from_topology(&metadata.mode) else {
        return;
    };
    if rollback.is_some() {
        return;
    }
    let is_host_client = route.is_host_client();

    // Only prepare input every send-interval.
    if !input_config.send_interval.is_zero() {
        let timer = send_timer
            .get_or_insert_with(|| Timer::new(input_config.send_interval, TimerMode::Repeating));
        timer.tick(real_time.delta());
        if !timer.is_finished() {
            return;
        }
    }

    #[cfg(not(feature = "prediction"))]
    if is_host_client {
        // if there is not prediction, no need to rebroadcast inputs
        return;
    }

    #[cfg(feature = "prediction")]
    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 = timeline.current_tick();
    let tick = current_tick + timeline.input_delay() as i32;
    // TODO: the number of messages should be in SharedConfig
    trace!(delayed_tick = ?tick, ?current_tick, "prepare_input_message");
    trace!(
        target: "lightyear_debug::input",
        kind = "prepare_input_message_start",
        schedule = "PostUpdate",
        sample_point = "PostUpdate",
        action = ?DebugName::type_name::<S::Action>(),
        local_tick = current_tick.0,
        input_tick = tick.0,
        is_host_client,
        "preparing 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?

    // Send redundant inputs so that if a packet is lost, we can still recover.
    // The size of the input bundle scales with `send_interval`.
    let mut num_ticks: u32 = ((input_config.send_interval.as_nanos() / tick_duration.as_nanos())
        + 1)
    .try_into()
    .unwrap();
    num_ticks *= input_config.packet_redundancy as u32;
    let mut message = InputMessage::<S>::new(tick);
    for (entity, input_buffer, pre_spawned, controlled_by) in input_buffer_query.iter() {
        if !route.accepts_local_target(controlled_by) {
            continue;
        }
        trace!(
            ?tick,
            ?entity,
            "Preparing input message with buffer: {:?}",
            input_buffer
        );

        let target = input_target(route, entity, pre_spawned);

        if let Some(state_sequence) = S::build_from_input_buffer(input_buffer, num_ticks, tick) {
            trace!(
                target: "lightyear_debug::input",
                kind = "prepare_input_message_target",
                schedule = "PostUpdate",
                sample_point = "PostUpdate",
                entity = ?entity,
                action = ?DebugName::type_name::<S::Action>(),
                local_tick = current_tick.0,
                input_tick = tick.0,
                num_ticks = num_ticks,
                buffer_len = input_buffer.len(),
                target = ?target,
                states = ?state_sequence,
                "added target data to input message"
            );
            message.inputs.push(PerTargetData {
                target,
                states: state_sequence,
            });
        }
    }

    // TODO: revisit this; maybe we should not send an empty message?
    // we send a message even when there are 0 inputs because that itself is information
    debug!(
        ?tick,
        ?num_ticks,
        ?is_host_client,
        "sending input message for {:?}: {}",
        DebugName::type_name::<S::Action>().shortname(),
        message
    );
    trace!(
        target: "lightyear_debug::input",
        kind = "prepare_input_message_finish",
        schedule = "PostUpdate",
        sample_point = "PostUpdate",
        action = ?DebugName::type_name::<S::Action>(),
        local_tick = current_tick.0,
        input_tick = tick.0,
        end_tick = tick.0,
        num_ticks = num_ticks,
        num_targets = message.inputs.len(),
        is_host_client,
        message = ?message,
        "prepared input 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
}

#[cfg(feature = "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,
    tick_duration: Res<TickDuration>,
    timeline: SyncedLocalTimeline,
    input_config: Res<InputConfig<S::Action>>,
    #[cfg(feature = "metrics")] mut input_metric_handles: ResMut<InputMetricHandles<S>>,
    metadata: Res<NetworkingMetadata>,
    last_confirmed_input: Res<LastConfirmedInput>,
    prediction_manager: Option<Res<PredictionManager>>,
    mut receivers: Query<&mut MessageReceiver<InputMessage<S>>>,
    mut predicted_query: Query<
        Option<&mut InputBuffer<S::Snapshot, S::Action>>,
        (Without<S::Marker>, Allow<PredictionDisable>),
    >,
    prespawned: Query<(Entity, &PreSpawned)>,
) {
    let Some(route) = InputRoute::from_topology(&metadata.mode) else {
        return;
    };
    // The host-client receives authoritative input directly through the server-side pipeline.
    if route.is_host_client() {
        return;
    }
    // Conventional clients only receive other players' inputs when the server is configured to
    // rebroadcast them. Direct P2P peers always exchange remote inputs with each other.
    if matches!(route, InputRoute::ClientServer { .. }) && !input_config.rebroadcast_inputs {
        return;
    }
    let Some(prediction_manager) = prediction_manager else {
        return;
    };
    let tick = timeline.tick();
    let mut received_relevant_input = false;
    match route {
        InputRoute::ClientServer { link, .. } => {
            if let Ok(mut receiver) = receivers.get_mut(link) {
                received_relevant_input |= receive_remote_player_input_messages_from_receiver::<S>(
                    &mut receiver,
                    &mut commands,
                    *tick_duration,
                    tick,
                    None,
                    &prediction_manager,
                    &mut predicted_query,
                    &prespawned,
                    #[cfg(feature = "metrics")]
                    &mut input_metric_handles,
                );
            }
        }
        InputRoute::P2P(links) => {
            for link in links {
                let Ok(mut receiver) = receivers.get_mut(*link) else {
                    continue;
                };
                received_relevant_input |= receive_remote_player_input_messages_from_receiver::<S>(
                    &mut receiver,
                    &mut commands,
                    *tick_duration,
                    tick,
                    Some(*link),
                    &prediction_manager,
                    &mut predicted_query,
                    &prespawned,
                    #[cfg(feature = "metrics")]
                    &mut input_metric_handles,
                );
            }
        }
    }

    if received_relevant_input {
        last_confirmed_input
            .received_any_messages
            .store(true, bevy_platform::sync::atomic::Ordering::Relaxed);
    }
}

#[cfg(feature = "prediction")]
fn receive_remote_player_input_messages_from_receiver<S: ActionStateSequence>(
    receiver: &mut MessageReceiver<InputMessage<S>>,
    commands: &mut Commands,
    tick_duration: TickDuration,
    tick: Tick,
    p2p_link: Option<Entity>,
    prediction_manager: &PredictionManager,
    predicted_query: &mut Query<
        Option<&mut InputBuffer<S::Snapshot, S::Action>>,
        (Without<S::Marker>, Allow<PredictionDisable>),
    >,
    prespawned: &Query<(Entity, &PreSpawned)>,
    #[cfg(feature = "metrics")] input_metric_handles: &mut InputMetricHandles<S>,
) -> bool {
    let mut received_relevant_input = false;
    receiver.receive().for_each(|message| {
        trace!(?message.end_tick, ?message, "received remote input message for action: {:?}", DebugName::type_name::<S::Action>());
        trace!(
            target: "lightyear_debug::input",
            kind = "remote_input_message_recv",
            schedule = "PreUpdate",
            sample_point = "PreUpdate",
            action = ?DebugName::type_name::<S::Action>(),
            local_tick = tick.0,
            end_tick = message.end_tick.0,
            num_targets = message.inputs.len(),
            rebroadcast = message.rebroadcast,
            message = ?message,
            "received remote player input message"
        );
        for target_data in message.inputs {
            let Some(entity) = (match target_data.target {
                InputTarget::Entity(entity) if p2p_link.is_none() => Some(entity),
                InputTarget::Entity(entity) => {
                    error!(
                        ?entity,
                        end_tick = ?message.end_tick,
                        "ignored P2P input target without a PreSpawned hash"
                    );
                    None
                }
                InputTarget::PreSpawned(hash) => {
                    prespawned.iter().find_map(|(entity, pre_spawned)| {
                        matches_prespawned_target(hash, p2p_link, pre_spawned).then_some(entity)
                    })
                }
            }) else {
                if message.rebroadcast {
                    debug!(
                        target = ?target_data.target,
                        end_tick = ?message.end_tick,
                        "ignored stale remote player input message for unmapped entity"
                    );
                } else if let Some(link) = p2p_link {
                    warn!(
                        ?link,
                        target = ?target_data.target,
                        "could not find a PreSpawned P2P input target owned by the sending Link"
                    );
                } else {
                    warn!("Could not find entity in entity_map for remote player input message {:?}", target_data.target);
                }
                continue;
            };
            debug!(
                ?tick, ?message.end_tick,
                "received remote client input message for entity: {:?}. Applying to diff buffer.",
                entity
            );
            trace!(
                target: "lightyear_debug::input",
                kind = "remote_input_target",
                schedule = "PreUpdate",
                sample_point = "PreUpdate",
                entity = ?entity,
                action = ?DebugName::type_name::<S::Action>(),
                local_tick = tick.0,
                end_tick = message.end_tick.0,
                target = ?target_data.target,
                states = ?target_data.states,
                "applying remote input target data"
            );
            // TODO: careful of entity collisions!
            let Ok(input_buffer) = predicted_query.get_mut(entity) else {
                if message.rebroadcast {
                    debug!(
                        ?entity,
                        ?target_data.states,
                        end_tick = ?message.end_tick,
                        "ignored stale remote player input message for unrecognized entity"
                    );
                } else {
                    error!(?entity, ?target_data.states, end_tick = ?message.end_tick, "received input message for unrecognized entity");
                }
                continue
            };
            trace!(predicted=?entity, end_tick = ?message.end_tick, "update action diff buffer for remote player PREDICTED using input message");

            // Adding a missing InputBuffer is deferred, so its Add observer has
            // not populated the metric cache by the time we record this message.
            #[cfg(feature = "metrics")]
            if input_buffer.is_none() {
                input_metric_handles.insert_entity(entity);
            }

            if let Some(mut input_buffer) = input_buffer {
                // do not parse the remote message our current Buffer end_tick is later than the message end_tick
                // this can happen if we receive multiple messages out of order.
                if input_buffer.last_remote_tick.is_some_and(|t| t >= message.end_tick) {
                    trace!("Ignoring input message because our current last_remote_tick {:?} is more recent than the remote_end_tick {:?}", input_buffer.last_remote_tick, message.end_tick);
                    trace!(
                        target: "lightyear_debug::input",
                        kind = "remote_input_ignored_stale",
                        schedule = "PreUpdate",
                        sample_point = "PreUpdate",
                        entity = ?entity,
                        action = ?DebugName::type_name::<S::Action>(),
                        local_tick = tick.0,
                        end_tick = message.end_tick.0,
                        last_remote_tick = ?input_buffer.last_remote_tick,
                        "ignored stale remote input message"
                    );
                    continue
                }
                received_relevant_input = true;
                update_buffer_from_remote_player_message::<S>(
                    target_data.states,
                    &mut input_buffer,
                    tick,
                    message.end_tick,
                    entity,
                    prediction_manager,
                    tick_duration,
                    #[cfg(feature = "metrics")]
                    &*input_metric_handles,
                );
            } else {
                // add the ActionState and InputBuffer if they are missing
                let mut input_buffer = InputBuffer::<S::Snapshot, S::Action>::default();
                received_relevant_input = true;
                update_buffer_from_remote_player_message::<S>(
                    target_data.states,
                    &mut input_buffer,
                    tick,
                    message.end_tick,
                    entity,
                    prediction_manager,
                    tick_duration,
                    #[cfg(feature = "metrics")]
                    &*input_metric_handles,
                );
                // Initialize ActionState from the latest buffered input rather
                // than from base_value(). When the remote player's end_tick is
                // in the past, get_action_state will call get(current_tick)
                // which returns None and leaves ActionState unchanged. If we
                // initialized to base_value() the player would simulate with
                // empty/released inputs until the buffer catches up.
                let mut action_state = S::State::base_value();
                if let Some(last) = input_buffer.get_last() {
                    S::from_snapshot(S::State::as_mut(&mut action_state), last);
                }
                commands.entity(entity).insert((input_buffer, action_state));
            };
        }
    });
    received_relevant_input
}

#[cfg(feature = "prediction")]
/// Update the LastConfirmedInput tick by looking at latest tick buffered in each InputBuffer.
///
/// This runs in PostUpdate, and not in PreUpdate, because we want the rollback to use the previous LastConfirmedInput tick.
/// For example if the last confirmed input was at tick 10, and we received an InputMessage with end_tick 14, we want to rollback to tick 10
/// (and not to tick 14) because we need to potentially re-apply inputs for ticks 11, 12, 13, 14.
fn update_last_confirmed_input<S: ActionStateSequence>(
    timeline: SyncedLocalTimeline,
    action_config: Res<InputConfig<S::Action>>,
    input_config: Res<InputTimelineConfig>,
    metadata: Res<NetworkingMetadata>,
    mut last_confirmed_input: ResMut<LastConfirmedInput>,
    predicted_query: Query<
        &InputBuffer<S::Snapshot, S::Action>,
        (Without<S::Marker>, Allow<PredictionDisable>),
    >,
) {
    let Some(route) = InputRoute::from_topology(&metadata.mode) else {
        return;
    };
    if matches!(route, InputRoute::ClientServer { .. }) && !action_config.rebroadcast_inputs {
        return;
    }
    let tick = timeline.tick();
    // Preserve the conventional lockstep behavior: input delay guarantees that the sole remote
    // endpoint has supplied the current tick. P2P must inspect the actual global frontier because
    // one slow peer can still be missing while the others are ready.
    if input_config.is_lockstep() && !matches!(metadata.mode, NetworkTopology::P2P { .. }) {
        last_confirmed_input.tick.set_if_lower(tick);
        return;
    }
    // TODO: how to handle multiple actions S?

    // find the earliest last_confirmed_tick for each client
    // The tracker was reset to a high tick, so the first real tick always becomes the minimum.
    let mut received_for_all_clients = true;
    predicted_query.iter().for_each(|buffer| {
        // if we received any messages, we update the LastConfirmedInput
        // (this is used to determine the last confirmed tick for each client)
        if let Some(end_tick) = buffer.last_remote_tick {
            last_confirmed_input.tick.set_if_lower(end_tick);
        } else {
            received_for_all_clients = false;
        }
    });
    // Several generic input plugins contribute sequentially in the same set. The tracker is reset
    // to true once in PreUpdate, so AND-ing preserves a missing stream reported by any input type.
    last_confirmed_input.received_for_all_clients &= received_for_all_clients;
    trace!(
        target: "lightyear_debug::input",
        kind = "last_confirmed_input",
        schedule = "PostUpdate",
        sample_point = "PostUpdate",
        action = ?DebugName::type_name::<S::Action>(),
        local_tick = tick.0,
        confirmed_tick = last_confirmed_input.tick.get().0,
        "updated LastConfirmedInput"
    );
}

#[cfg(feature = "prediction")]
fn update_buffer_from_remote_player_message<S: ActionStateSequence>(
    sequence: S,
    input_buffer: &mut InputBuffer<S::Snapshot, S::Action>,
    tick: Tick,
    end_tick: Tick,
    entity: Entity,
    prediction_manager: &PredictionManager,
    tick_duration: TickDuration,
    #[cfg(feature = "metrics")] input_metric_handles: &InputMetricHandles<S>,
) {
    // we don't need to update the ActionState here because:
    // - if it's in the future, we will fetch the ActionState from the buffer in the `get_action_state` system when
    //   we reach that tick.
    // - if it's in the past, we will start a rollback and fetch the correct ActionState from the buffer in the
    //   `get_action_state` system
    if let Some(mismatch) = sequence.update_buffer(input_buffer, end_tick, tick_duration.0) {
        trace!(
            target: "lightyear_debug::input",
            kind = "remote_input_buffer_update",
            schedule = "PreUpdate",
            sample_point = "PreUpdate",
            entity = ?entity,
            action = ?DebugName::type_name::<S::Action>(),
            local_tick = tick.0,
            end_tick = end_tick.0,
            mismatch_tick = mismatch.0,
            buffer_len = input_buffer.len(),
            last_remote_tick = ?input_buffer.last_remote_tick,
            input_buffer = %input_buffer,
            "updated remote input buffer with mismatch"
        );
        // find the earliest mismatch tick across all clients
        // NOTE: if the mismatch tick is more recent than our current tick, then it's not a mismatch!
        //  it just means that we are receiving a remote tick in advance of simulating that tick.
        //  (for example in lockstep mode, we should have all player inputs for tick T before simulating tick T,
        //  so we will receive those inputs in advance)
        if let RollbackMode::Check = prediction_manager.rollback_policy.input
            && mismatch <= tick
        {
            debug!(
                ?entity,
                ?tick,
                ?end_tick,
                ?mismatch,
                "Mismatch detected for remote player input message!",
            );
            prediction_manager
                .earliest_mismatch_input
                .has_mismatches
                .store(true, bevy_platform::sync::atomic::Ordering::Relaxed);
            prediction_manager
                .earliest_mismatch_input
                .tick
                .set_if_lower(mismatch);
        }

        #[cfg(feature = "metrics")]
        {
            input_metric_handles.remote_player_receive().increment(1);
            let margin = input_buffer.last_remote_tick.unwrap() - tick;
            input_metric_handles
                .remote_player_buffer_margin(entity)
                .set(margin as f64);
            input_metric_handles
                .remote_player_buffer_size(entity)
                .set(input_buffer.len() as f64);
        }
    };
    trace!(
        target: "lightyear_debug::input",
        kind = "remote_input_buffer_state",
        schedule = "PreUpdate",
        sample_point = "PreUpdate",
        entity = ?entity,
        action = ?DebugName::type_name::<S::Action>(),
        local_tick = tick.0,
        end_tick = end_tick.0,
        buffer_len = input_buffer.len(),
        last_remote_tick = ?input_buffer.last_remote_tick,
        input_buffer = %input_buffer,
        "remote input buffer state after message"
    );
}

/// Drain the messages from the buffer and send them to the server
#[cfg_attr(
    not(feature = "interpolation"),
    expect(
        unused_mut,
        reason = "InputMessage is mutated only when interpolation is enabled"
    )
)]
fn send_input_messages<S: ActionStateSequence>(
    input_config: Res<InputConfig<S::Action>>,
    _local_timeline: SyncedLocalTimeline,
    #[cfg(feature = "interpolation")] fixed_time: Res<Time<Fixed>>,
    metadata: Res<NetworkingMetadata>,
    mut message_buffer: ResMut<MessageBuffer<S>>,
    mut senders: Query<&mut MessageSender<InputMessage<S>>>,
    mut multi_sender: MultiMessageSender,
    #[cfg(feature = "interpolation")] interpolation_timeline: Res<InterpolationTimeline>,
) {
    // An inactive topology retains the buffered messages, matching the old `Single<Client>`
    // parameter's skip-on-no-match behavior across disconnect/reconnect boundaries.
    let Some(route) = InputRoute::from_topology(&metadata.mode) else {
        return;
    };
    let is_host_client = route.is_host_client();

    #[cfg(not(feature = "prediction"))]
    if is_host_client {
        message_buffer.0.clear();
        return;
    }
    #[cfg(feature = "prediction")]
    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()
    );
    trace!(
        target: "lightyear_debug::input",
        kind = "send_input_messages",
        schedule = "PostUpdate",
        sample_point = "PostUpdate",
        action = ?DebugName::type_name::<S::Action>(),
        num_messages = message_buffer.0.len(),
        is_host_client,
        "sending buffered input messages"
    );

    #[cfg(feature = "interpolation")]
    let interpolation_delay = match route {
        InputRoute::ClientServer { .. } => {
            if !interpolation_timeline.is_synced() {
                return;
            }
            // NOTE: this can be negative because of input-delay!
            let mut delay = _local_timeline.instant(&fixed_time) - interpolation_timeline.now();
            if delay.is_negative() {
                delay = TickDelta::from(Tick(0));
            }
            Some(InterpolationDelay {
                delay: delay.into(),
            })
        }
        // Input-only P2P has no interpolation timeline or server lag-compensation target.
        InputRoute::P2P(_) => None,
    };

    match route {
        InputRoute::ClientServer { link, .. } => {
            let Ok(mut sender) = senders.get_mut(link) else {
                return;
            };
            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 = interpolation_delay;
                }
                sender.send::<InputChannel>(message);
            }
        }
        InputRoute::P2P(links) => {
            let Some(links) = unique_p2p_links(links) else {
                error!("cached P2P Link entities must be unique");
                message_buffer.0.clear();
                return;
            };
            for mut message in message_buffer.0.drain(..) {
                #[cfg(feature = "interpolation")]
                if input_config.lag_compensation {
                    message.interpolation_delay = interpolation_delay;
                }
                if let Err(error) =
                    multi_sender.send::<InputMessage<S>, InputChannel>(&message, links)
                {
                    error!(%error, "failed to send input message to P2P Links");
                }
            }
        }
    }
}

/// View the cached P2P Link slice as Bevy's allocation-free entity-set type.
///
/// `NetworkingMetadata` builds this slice from distinct query entities. The explicit check keeps
/// this conversion sound even if user code mutates the public cached metadata.
fn unique_p2p_links(links: &[Entity]) -> Option<&UniqueEntitySlice> {
    let has_duplicates = links
        .iter()
        .enumerate()
        .any(|(index, link)| links[index + 1..].contains(link));
    if has_duplicates {
        return None;
    }
    // SAFETY: the pairwise check above proves that every Entity in the slice is unique.
    Some(unsafe { UniqueEntitySlice::from_slice_unchecked(links) })
}

/// In case the client tick changes suddenly, we also update the InputBuffer accordingly
fn receive_local_timeline_shift<S: ActionStateSequence>(
    trigger: On<LocalTimelineShift>,
    mut message_buffer: ResMut<MessageBuffer<S>>,
    mut input_buffer_query: Query<
        &mut InputBuffer<S::Snapshot, S::Action>,
        Allow<PredictionDisable>,
    >,
) {
    let delta = trigger.delta;
    for mut input_buffer in input_buffer_query.iter_mut() {
        let had_start_tick = input_buffer.start_tick.is_some();
        shift_input_buffer_ticks(&mut input_buffer, delta);
        if had_start_tick {
            debug!(
                "Receive local timeline shift {:?}. 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;
    }
}

fn shift_input_buffer_ticks<S, A>(input_buffer: &mut InputBuffer<S, A>, delta: i32) {
    if let Some(start_tick) = input_buffer.start_tick {
        input_buffer.start_tick = Some(start_tick + delta);
    }
    if let Some(last_remote_tick) = input_buffer.last_remote_tick {
        input_buffer.last_remote_tick = Some(last_remote_tick + delta);
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use lightyear_replication::prelude::Lifetime;

    #[test]
    fn input_route_uses_link_ownership_only_for_client_server() {
        let mut world = World::new();
        let client = world.spawn_empty().id();
        let other = world.spawn_empty().id();
        let owned_by_client = ControlledBy {
            owner: client,
            lifetime: Lifetime::SessionBased,
        };
        let owned_by_other = ControlledBy {
            owner: other,
            lifetime: Lifetime::SessionBased,
        };

        let client_server = NetworkTopology::Client(client);
        let client_server_route = InputRoute::from_topology(&client_server).unwrap();
        assert!(client_server_route.accepts_local_target(None));
        assert!(client_server_route.accepts_local_target(Some(&owned_by_client)));
        assert!(!client_server_route.accepts_local_target(Some(&owned_by_other)));

        let p2p = NetworkTopology::P2P {
            connected: [client, other].into_iter().collect(),
            declared_links: 2,
        };
        let p2p_route = InputRoute::from_topology(&p2p).unwrap();
        assert!(p2p_route.accepts_local_target(None));
        assert!(p2p_route.accepts_local_target(Some(&owned_by_client)));
        assert!(p2p_route.accepts_local_target(Some(&owned_by_other)));
    }

    #[test]
    fn p2p_input_targets_use_prespawned_hashes() {
        let mut world = World::new();
        let link = world.spawn_empty().id();
        let target = world.spawn_empty().id();
        let topology = NetworkTopology::P2P {
            connected: [link].into_iter().collect(),
            declared_links: 1,
        };
        let route = InputRoute::from_topology(&topology).unwrap();
        let prespawned = PreSpawned::new(0xCAFE);

        assert_eq!(
            input_target(route, target, Some(&prespawned)),
            InputTarget::PreSpawned(0xCAFE)
        );
    }

    #[test]
    #[should_panic(expected = "must have a PreSpawned component with a resolved hash")]
    fn p2p_input_targets_reject_unmapped_entities() {
        let mut world = World::new();
        let link = world.spawn_empty().id();
        let target = world.spawn_empty().id();
        let topology = NetworkTopology::P2P {
            connected: [link].into_iter().collect(),
            declared_links: 1,
        };
        let route = InputRoute::from_topology(&topology).unwrap();

        let _ = input_target(route, target, None);
    }

    #[test]
    fn client_server_input_targets_still_allow_entities() {
        let mut world = World::new();
        let link = world.spawn_empty().id();
        let target = world.spawn_empty().id();
        let topology = NetworkTopology::Client(link);
        let route = InputRoute::from_topology(&topology).unwrap();

        assert_eq!(
            input_target(route, target, None),
            InputTarget::Entity(target)
        );
    }

    #[test]
    fn p2p_input_targets_are_scoped_to_the_sending_link() {
        let mut world = World::new();
        let owner = world.spawn_empty().id();
        let other = world.spawn_empty().id();
        let pre_spawned = PreSpawned::new(0xCAFE).for_receiver(owner);

        assert!(matches_prespawned_target(0xCAFE, Some(owner), &pre_spawned));
        assert!(!matches_prespawned_target(
            0xCAFE,
            Some(other),
            &pre_spawned
        ));
        assert!(matches_prespawned_target(0xCAFE, None, &pre_spawned));
    }

    #[test]
    fn input_history_depth_covers_prediction_rollback_window() {
        assert_eq!(input_history_depth(None), HISTORY_DEPTH);

        let input_config = InputTimelineConfig::default();

        let mut manager = PredictionManager {
            rollback_policy: RollbackPolicy {
                max_rollback_ticks: 100,
                ..Default::default()
            },
            ..Default::default()
        };
        assert_eq!(input_history_depth(Some((&manager, &input_config))), 101);

        manager.rollback_policy.max_rollback_ticks = 5;
        assert_eq!(
            input_history_depth(Some((&manager, &input_config))),
            HISTORY_DEPTH
        );
    }
}