//! Executor for the per-peer control machine's [`PeerAction`]s.
//!
//! The per-peer FSM in [`crate::peer::machine`] is a sans-IO reducer: it decides
//! *what* must happen and returns a `Vec<PeerAction>`; this module is the *doing*
//! half — the thin driver that maps each action onto the exact shell call it
//! stands for (`build_msg2` + `transport.send`, `promote_connection`,
//! `remove_active_peer`, `index_allocator.free`, `note_link_dead`, …).
//!
//! ## Progressive cutover
//!
//! The executor is wired incrementally. Live today: the inbound establish
//! (`handle_msg1` → `step(InboundMsg1)`), the outbound msg2 promote
//! (`handle_msg2` looks up the dial-persisted machine), the connectionless
//! outbound msg1 send (`SendHandshake` with `their_index == None` →
//! `send_stored_msg1`, driven from `initiate_connection`), the
//! connection-oriented dial (`OpenTransport` performs the non-blocking
//! `transport.connect`; `TransportConnected` drives the connect-resolution msg1
//! send from `poll_pending_connects`), the rekey cadence (`check_rekey` →
//! `route_rekey_cadence` → `RekeyConsume`, driving the `SwapSendState` and
//! `CompleteDrain` arms), and the liveness reap (`route_link_dead` →
//! `LinkDeadSuspected`, driving `InvalidateSendState` → `remove_active_peer`).
//!
//! The genuine inert stubs remaining are `SendRekey`, `SendLinkMessage`, and
//! the connected-UDP arms. `RegisterDecryptSession` is a deliberate no-op —
//! see its arm for the note.
//!
//! The timer arms (`SetTimer`/`CancelTimer`) populate/clear the per-peer timer
//! store (`peer_timers`). The `HandshakeRetransmit` and `HandshakeTimeout`
//! deadlines are read and fired by `drive_peer_timers` (the handshake resend +
//! reap home). The rekey/liveness kinds are still SHADOW — driven by their own
//! shell drivers — so populating them stays behavior-neutral.

use crate::PeerIdentity;
use crate::node::Node;
use crate::node::reject::{HandshakeReject, RejectReason};
use crate::peer::machine::{LostKind, PeerAction, PeerEvent};
use crate::proto::fmp::PromotionResult;
use crate::proto::fmp::wire::build_msg2;
use crate::transport::{LinkId, TransportAddr, TransportId};
use crate::utils::index::SessionIndex;
use std::collections::VecDeque;
use tracing::{debug, trace, warn};

/// Ambient shell facts a [`PeerAction`] executor needs that the machine's
/// runtime-agnostic action payloads deliberately omit (verified identity,
/// transport target, the msg2 framing indices, the promotion timestamp).
///
/// Unlike a machine event/action payload this is **executor-side**, so it may
/// hold real values resolved from the wire context (cf. `handle_msg1`'s
/// `wire`/`packet` locals and `promote_connection`'s ambient args). It is
/// built fresh per driven step by the caller at cutover time.
#[allow(dead_code)]
pub(in crate::node) struct PeerActionCtx {
    /// The authenticated peer identity: `PromoteToActive` /
    /// `InvalidateSendState` resolve their `NodeAddr` from this.
    pub(in crate::node) verified_identity: PeerIdentity,
    /// The transport the exchange is happening over (msg2 send target, decrypt
    /// cache-key transport half).
    pub(in crate::node) transport_id: TransportId,
    /// The peer's wire address (msg2 send target).
    pub(in crate::node) remote_addr: TransportAddr,
    /// Our session index for this exchange (msg2 framing sender_idx).
    pub(in crate::node) our_index: Option<SessionIndex>,
    /// The peer's session index for this exchange (msg2 framing
    /// receiver_idx).
    pub(in crate::node) their_index: Option<SessionIndex>,
    /// The wire timestamp driving this step (promotion ts / loss-report clock).
    pub(in crate::node) now_ms: u64,
    /// Establish direction for this exchange. Discriminates the
    /// `PromoteToActive` failure cleanup: the pre-refactor inbound
    /// (`handle_msg1`) and outbound (`handle_msg2`) promote-Err arms were NOT
    /// byte-identical, so the executor must reproduce each. `false` = inbound
    /// (drop link + reverse map + free index), `true` = outbound (record the
    /// reject only; leave the dead link/`addr_to_link` for the stale-connection
    /// reaper, matching old `handle_msg2`).
    pub(in crate::node) is_outbound: bool,
}

impl Node {
    /// Advance the machine for `link` by one event and execute the resulting
    /// actions.
    ///
    /// The borrow structure the whole seam turns on: the machine
    /// needs `&mut IndexAllocator` as a synchronous capability *while it is
    /// itself borrowed mutably out of `peer_machines`*. `peer_machines` and
    /// `index_allocator` are **distinct `Node` fields**, so the collect below is
    /// a disjoint two-field borrow the checker accepts; once the actions are
    /// collected both borrows drop and the executor runs against `&mut self`.
    pub(in crate::node) async fn advance_peer_machine(
        &mut self,
        link: LinkId,
        event: PeerEvent,
        now: u64,
        ambient: &PeerActionCtx,
    ) {
        let actions = match self.peer_machines.get_mut(&link) {
            // Disjoint field borrow: `self.peer_machines` (the map entry) and
            // `self.index_allocator` (the capability) are separate fields.
            Some(machine) => machine.step(event, now, &mut self.index_allocator),
            None => return,
        };
        self.execute_peer_actions(link, ambient, actions).await;
    }

    /// Map each [`PeerAction`] onto its shell call.
    ///
    /// `PromoteToActive` feeds its [`PromotionResult`](crate::proto::fmp::PromotionResult)
    /// back into the machine and appends the follow-up actions to the same
    /// worklist — a queue rather than self-recursion so the async executor stays a
    /// single flat future (no boxing) and the emitted order is preserved (the
    /// establish sequences always end in `PromoteToActive`, so its follow-ups run
    /// after any siblings).
    pub(in crate::node) async fn execute_peer_actions(
        &mut self,
        link: LinkId,
        ambient: &PeerActionCtx,
        actions: Vec<PeerAction>,
    ) {
        let mut queue: VecDeque<PeerAction> = actions.into();
        while let Some(action) = queue.pop_front() {
            match action {
                PeerAction::OpenTransport {
                    transport_id,
                    remote_addr,
                } => {
                    // Outbound connection-oriented dial. `initiate_connection`'s
                    // oriented branch drove the machine to `Connecting`, which
                    // emitted this action. Perform the non-blocking
                    // `transport.connect` and, on success, push the
                    // `PendingConnect` for `poll_pending_connects` to resolve. On
                    // connect error, tear down the dial-window state (link,
                    // reverse map, control machine) and abort the queue — the
                    // executor-local mirror of the old inline
                    // `initiate_connection` connect+push.
                    if let Some(transport) = self.transports.get(&transport_id) {
                        match transport.connect(&remote_addr).await {
                            Ok(()) => {
                                debug!(
                                    transport_id = %transport_id,
                                    remote_addr = %remote_addr,
                                    link_id = %link,
                                    "Transport connect initiated (non-blocking)"
                                );
                                self.peering
                                    .pending_connects
                                    .push(crate::node::PendingConnect {
                                        link_id: link,
                                        transport_id,
                                        remote_addr,
                                        peer_identity: ambient.verified_identity,
                                    });
                            }
                            Err(_e) => {
                                self.links.remove(&link);
                                self.addr_to_link.remove(&(transport_id, remote_addr));
                                self.remove_peer_machine(link);
                                return;
                            }
                        }
                    }
                }
                PeerAction::SendHandshake { bytes } => {
                    // Two outbound directions share this action, discriminated by
                    // `their_index`:
                    //   msg2 (`their_index == Some`): the machine payload is the
                    //   UNFRAMED Noise msg2; frame it with our/their index
                    //   (`build_msg2`) and send.
                    //   msg1 (`their_index == None`): a fresh outbound handshake;
                    //   the machine's empty payload is ignored — the shell already
                    //   allocated the index, ran the Noise leaf, and armed the
                    //   wire at dial (`prepare_outbound_msg1`); this just sends the
                    //   stored wire (see `send_stored_msg1`).
                    if let (Some(sender_idx), Some(receiver_idx)) =
                        (ambient.our_index, ambient.their_index)
                    {
                        let frame = build_msg2(sender_idx, receiver_idx, &bytes);
                        // Surface the send Result. A missing transport skips
                        // the send and continues (mirrors `handle_msg1`'s
                        // `if let Some(transport)` guard); a send *error* runs the
                        // pre-refactor msg2-send-failure cleanup (`handle_msg1`
                        // L494-503) and ABORTS the remaining queue so the queued
                        // `PromoteToActive` never runs.
                        let send_err = match self.transports.get(&ambient.transport_id) {
                            Some(transport) => {
                                transport.send(&ambient.remote_addr, &frame).await.err()
                            }
                            None => None,
                        };
                        if let Some(e) = send_err {
                            // Restored pre-refactor msg2-send-failure warn!
                            // (`handle_msg1` L665): the send error text is surfaced
                            // at the executor point where the failure is now handled.
                            warn!(link_id = %link, error = %e, "Failed to send msg2");
                            self.links.remove(&link);
                            self.addr_to_link
                                .remove(&(ambient.transport_id, ambient.remote_addr.clone()));
                            if let Some(idx) = ambient.our_index {
                                let _ = self.index_allocator.free(idx);
                            }
                            self.remove_peer_machine(link);
                            self.stats_mut()
                                .record_reject(RejectReason::Handshake(HandshakeReject::BadState));
                            return;
                        }
                    } else {
                        // msg1: the shell already allocated the index, ran the
                        // Noise leaf, and armed the wire on the connection at dial
                        // (`prepare_outbound_msg1`); send the stored wire. The
                        // machine's empty payload is ignored.
                        let _ = bytes;
                        self.send_stored_msg1(
                            link,
                            ambient.transport_id,
                            &ambient.remote_addr,
                            ambient.now_ms,
                        )
                        .await;
                    }
                }
                PeerAction::SendRekey { .. } => {
                    // Rekey msg2 framing (`build_msg2(our_new_index, …)`,
                    // `handshake.rs:365`) + send. Rekey fold is not yet wired.
                }
                PeerAction::SendLinkMessage { .. } => {
                    // Encrypt + send a link-control frame (heartbeat / filter
                    // / tree / disconnect). Data-plane-owned; not yet wired.
                }
                PeerAction::PromoteToActive { link: promote_link } => {
                    // Ambient supplies the verified identity + promotion ts
                    // that `promote_connection` needs (resolved from the wire ctx).
                    match self.promote_connection(
                        promote_link,
                        ambient.verified_identity,
                        ambient.now_ms,
                    ) {
                        Ok(result) => {
                            // The decrypt-worker registration relocated
                            // OUT of `promote_connection` into THIS single executor
                            // arm — the one live caller of `promote_connection` (both
                            // the inbound `handle_msg1` and outbound `handle_msg2`
                            // net-new establish paths reach it here). Register iff the
                            // promotion actually created or replaced a peer
                            // (`Promoted | CrossConnectionWon`), NEVER on
                            // `CrossConnectionLost`. Run synchronously right after
                            // `promote_connection` returns, before feeding
                            // `PromotionResolved` and before any await — the exact
                            // synchronous point (and Promoted/Won gating) of the
                            // pre-refactor in-`promote_connection` call. No-op when
                            // the worker pool isn't spawned (`register_...` early-
                            // returns), so the direct `promote_connection` test
                            // callers (which bypass this executor) are unaffected.
                            #[cfg(unix)]
                            match result {
                                PromotionResult::Promoted(node_addr)
                                | PromotionResult::CrossConnectionWon { node_addr, .. } => {
                                    self.register_decrypt_worker_session(&node_addr);
                                }
                                PromotionResult::CrossConnectionLost { .. } => {}
                            }

                            // Feed the outcome back into the machine and fold the
                            // follow-up actions (RegisterDecryptSession — now a
                            // redundant no-op, see its arm — and the cross-conn index
                            // frees) into the worklist. Disjoint field borrow again.
                            let follow = match self.peer_machines.get_mut(&promote_link) {
                                Some(machine) => machine.step(
                                    PeerEvent::PromotionResolved { result },
                                    ambient.now_ms,
                                    &mut self.index_allocator,
                                ),
                                None => Vec::new(),
                            };
                            queue.extend(follow);

                            // Defensive cross-connection loser-link surgery.
                            // LINK-ONLY: close the losing transport connection, drop
                            // its link, and re-point `addr_to_link`, reproducing the
                            // pre-refactor inline `handle_msg2`/`handle_msg1` per-arm
                            // order EXACTLY. The index-plane frees/unregisters are
                            // owned by the machine's `PromotionResolved{Won/Lost}`
                            // follow-up (queued just above), so NOTHING here touches
                            // an index — no double-free.
                            //
                            // UNREACHABLE on every current driven path: the inbound
                            // and outbound net-new establish arms only route to the
                            // machine when no promoted peer exists for the node_addr
                            // (and `RestartThenPromote` removes the old peer first),
                            // so `promote_connection` always returns `Promoted`. The
                            // `debug_assert!(false, ..)` catches any future path that
                            // drives a cross-connection through the executor without
                            // the matching send-state handling.
                            match result {
                                PromotionResult::CrossConnectionWon { loser_link_id, .. } => {
                                    debug_assert!(
                                        false,
                                        "executor CrossConnectionWon is unreachable on \
                                         driven net-new establish paths"
                                    );
                                    // Close the losing transport connection (no-op for
                                    // connectionless) via the LOSER link's own
                                    // transport/addr, then drop the losing link.
                                    if let Some(loser_link) = self.links.get(&loser_link_id) {
                                        let loser_tid = loser_link.transport_id();
                                        let loser_addr = loser_link.remote_addr().clone();
                                        if let Some(transport) = self.transports.get(&loser_tid) {
                                            transport.close_connection(&loser_addr).await;
                                        }
                                    }
                                    self.remove_link(&loser_link_id);
                                    // Point `addr_to_link` at the winning (current)
                                    // link.
                                    self.addr_to_link.insert(
                                        (ambient.transport_id, ambient.remote_addr.clone()),
                                        promote_link,
                                    );
                                }
                                PromotionResult::CrossConnectionLost { winner_link_id } => {
                                    debug_assert!(
                                        false,
                                        "executor CrossConnectionLost is unreachable on \
                                         driven net-new establish paths"
                                    );
                                    // Close this (losing) connection, drop its link,
                                    // and restore `addr_to_link` to the winner.
                                    if let Some(transport) =
                                        self.transports.get(&ambient.transport_id)
                                    {
                                        transport.close_connection(&ambient.remote_addr).await;
                                    }
                                    self.remove_link(&promote_link);
                                    self.addr_to_link.insert(
                                        (ambient.transport_id, ambient.remote_addr.clone()),
                                        winner_link_id,
                                    );
                                }
                                PromotionResult::Promoted(_) => {}
                            }
                        }
                        Err(e) => {
                            // Promotion failed. `promote_connection` already
                            // removed `connections[link]` and (on error) handled its
                            // own index internally. The pre-refactor inbound and
                            // outbound promote-Err arms were NOT byte-identical, so
                            // discriminate on `ambient.is_outbound`. The queue is
                            // drained (PromoteToActive is the last establish action),
                            // so no explicit abort.
                            if ambient.is_outbound {
                                // OLD outbound (`handle_msg2` promote-Err): warn +
                                // record_reject ONLY. NO `remove_link`, NO
                                // `index_allocator.free`, NO `addr_to_link` removal —
                                // the dead link/addr_to_link/pending_outbound were
                                // left for the 30s stale-connection reaper
                                // (`promote_connection` already handled
                                // `connections[link]`/its index on error). Restored
                                // pre-refactor outbound warn! ("Failed to promote
                                // connection").
                                //
                                // The outbound machine was persisted at dial; it is
                                // additive state that did not exist pre-refactor, so
                                // removing it on promote failure is neutral vs old and
                                // prevents a leak.
                                warn!(
                                    target: "fips::node::handlers::handshake",
                                    link_id = %promote_link,
                                    error = %e,
                                    "Failed to promote connection"
                                );
                                self.stats_mut().record_reject(RejectReason::Handshake(
                                    HandshakeReject::BadState,
                                ));
                                self.remove_peer_machine(promote_link);
                            } else {
                                // OLD inbound (`handle_msg1` L587-591): drop the link
                                // + reverse map, free our index, discard the machine,
                                // and record the reject. Restored pre-refactor inbound
                                // promote-failure warn! (`handle_msg1` L757).
                                warn!(
                                    target: "fips::node::handlers::handshake",
                                    link_id = %promote_link,
                                    error = %e,
                                    "Failed to promote inbound connection"
                                );
                                self.remove_link(&promote_link);
                                if let Some(idx) = ambient.our_index {
                                    let _ = self.index_allocator.free(idx);
                                }
                                self.remove_peer_machine(promote_link);
                                self.stats_mut().record_reject(RejectReason::Handshake(
                                    HandshakeReject::BadState,
                                ));
                            }
                        }
                    }
                }
                PeerAction::ResolveCrossConnection { .. } => {
                    // A decision token, not an effect: the outbound msg2
                    // handler intercepts it and runs the inline swap/keep
                    // resolution itself, so it must never reach the executor.
                    debug_assert!(
                        false,
                        "ResolveCrossConnection is intercepted by the msg2 \
                         handler and must never reach the executor"
                    );
                }
                PeerAction::SwapSendState { .. } => {
                    // Initiator cutover: the live authoritative rekey-cadence
                    // path, routed here from `check_rekey` via
                    // `route_rekey_cadence` → `PeerEvent::RekeyConsume`; the
                    // inline body survives only as `cutover_peer_inline`, a
                    // debug-assert release fallback. `addr` is resolved
                    // from the ambient verified identity (as `InvalidateSendState`
                    // does). The decrypt re-register folds HERE, gated on
                    // `did_cutover` — the generic `RegisterDecryptSession` arm stays a
                    // no-op so a promote never double-registers.
                    let node_addr = *ambient.verified_identity.node_addr();
                    let did_cutover = if let Some(peer) = self.peers.get_mut(&node_addr) {
                        if let Some(_old_our_index) = peer.cutover_to_new_session() {
                            // New index was pre-registered in peers_by_index
                            // during msg2 handling (handshake.rs).
                            debug_assert!(
                                peer.transport_id().is_some()
                                    && peer.our_index().is_some()
                                    && self.peers_by_index.contains_key(&(
                                        peer.transport_id().unwrap(),
                                        peer.our_index().unwrap().as_u32()
                                    )),
                                "peers_by_index should contain pre-registered new index after cutover"
                            );
                            debug!(
                                // Pin the target to the pre-refactor module: this
                                // cutover log relocated from handlers/rekey.rs into
                                // the executor, but operators (and the test harness)
                                // filter it under fips::node::handlers::rekey. Keeping
                                // the target preserves the observable log contract.
                                target: "fips::node::handlers::rekey",
                                peer = %self.peer_display_name(&node_addr),
                                "Rekey cutover complete (initiator), K-bit flipped"
                            );
                            true
                        } else {
                            false
                        }
                    } else {
                        false
                    };
                    // Re-register the new session with the decrypt worker — the
                    // cache_key (transport_id, our_index) just changed, so the
                    // old worker entry is stale and every packet on the new
                    // session would miss the worker's HashMap lookup.
                    #[cfg(unix)]
                    if did_cutover {
                        self.register_decrypt_worker_session(&node_addr);
                    }
                    #[cfg(not(unix))]
                    let _ = did_cutover;
                }
                PeerAction::CompleteDrain { peer: node_addr } => {
                    // Initiator drain completion: the live authoritative
                    // rekey-cadence path, routed here from `check_rekey` via
                    // `route_rekey_cadence` → `PeerEvent::RekeyConsume`; the
                    // inline body survives only as `drain_peer_inline`, a
                    // debug-assert release fallback. Extract the real previous
                    // index + transport_id under the peer borrow, drop the
                    // borrow, then run the cache_key cleanup (which takes
                    // &mut self for unregister_decrypt_worker_session).
                    let drained = self.peers.get_mut(&node_addr).and_then(|peer| {
                        peer.complete_drain().map(|idx| (idx, peer.transport_id()))
                    });
                    if let Some((old_our_index, transport_id)) = drained {
                        if let Some(tid) = transport_id {
                            let cache_key = (tid, old_our_index.as_u32());
                            self.peers_by_index.remove(&cache_key);
                            #[cfg(unix)]
                            self.unregister_decrypt_worker_session(cache_key);
                        }
                        let _ = self.index_allocator.free(old_our_index);
                        trace!(
                            // Pin to the pre-refactor module (see the cutover log
                            // above) so the relocated drain log stays visible under
                            // the operator's fips::node::handlers::rekey filter.
                            target: "fips::node::handlers::rekey",
                            peer = %self.peer_display_name(&node_addr),
                            old_index = %old_our_index,
                            "Drain complete, previous session erased"
                        );
                    }
                }
                PeerAction::InvalidateSendState => {
                    // The FULL teardown. `remove_active_peer`
                    // (`dispatch.rs:107`) frees the four index slots
                    // (current/rekey/pending/previous), drops `peers_by_index`,
                    // unregisters the decrypt worker, removes the FSP `sessions`
                    // entry and `pending_tun_packets`. The machine emits NO
                    // `FreeIndex` for those slots, so there is no double-free.
                    self.remove_active_peer(ambient.verified_identity.node_addr());
                }
                PeerAction::RegisterDecryptSession { index } => {
                    let _ = index;
                    // No-op by design. The decrypt-worker
                    // registration relocated into the `PromoteToActive` Ok arm above, gated on
                    // the returned `PromotionResult`, so it runs once per live
                    // promote (Promoted/Won) at the pre-refactor synchronous point.
                    // This machine-emitted action is now redundant with that arm;
                    // kept as an inert no-op (rather than removing the emission) so
                    // the machine's action sequence and its unit tests stay
                    // unchanged. The keyed-by-NodeAddr register does not need the
                    // machine's `index` payload.
                }
                PeerAction::UnregisterDecryptSession { index } => {
                    // Executor supplies `transport_id` from ambient; keyed by
                    // (tid, index) like `remove_active_peer` / the rekey drain path.
                    #[cfg(unix)]
                    self.unregister_decrypt_worker_session((ambient.transport_id, index.as_u32()));
                    #[cfg(not(unix))]
                    let _ = index;
                }
                PeerAction::FreeIndex { index } => {
                    let _ = self.index_allocator.free(index);
                }
                PeerAction::ActivateConnectedUdp | PeerAction::TeardownConnectedUdp => {
                    // Connected-UDP plane ownership (`connected_udp.rs`).
                }
                PeerAction::SetTimer { kind, at_ms } => {
                    // Populate the per-peer timer store (overwrite = reschedule).
                    // The `HandshakeRetransmit` and `HandshakeTimeout` deadlines
                    // are read + fired by `drive_peer_timers`. Rekey/liveness kinds
                    // are still SHADOW here — they keep their own shell drivers —
                    // so populating them stays behavior-neutral.
                    self.peer_timers
                        .entry(link)
                        .or_default()
                        .insert(kind, at_ms);
                }
                PeerAction::CancelTimer { kind } => {
                    if let Some(timers) = self.peer_timers.get_mut(&link) {
                        timers.remove(&kind);
                    }
                }
                PeerAction::ReportLost { peer, kind } => {
                    // The single loss token, routed to the reconciler reflex the
                    // `kind` names: an un-promoted handshake attempt takes the
                    // connected-guarded `note_handshake_timeout` (`driver.rs:28`),
                    // an established peer's link-death takes the unconditional
                    // `note_link_dead` (`driver.rs:48`).
                    match kind {
                        LostKind::HandshakeTimeout => {
                            self.note_handshake_timeout(peer, ambient.now_ms);
                        }
                        LostKind::LinkDead => {
                            self.note_link_dead(peer, ambient.now_ms);
                        }
                    }
                }
            }
        }
    }
}
