nym/nym-node/src/node/lp/data/handler/pipeline/mixing_node.rs

// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0

//! Pipeline for nodes operating purely as mixnodes (no client forwarding).
//!

use std::{sync::Arc, time::Instant};

use nym_lp_data::{
    AddressedTimedData, PipelinePayload, TimedData, TimedPayload,
    common::traits::{
        Framing, FramingUnwrap, Transport, TransportUnwrap, WireUnwrappingPipeline,
        WireWrappingPipeline,
    },
    nymnodes::traits::NymNodeProcessingPipeline,
    packet::{EncryptedLpPacket, LpFrame, LpHeader, MalformedLpPacketError, frame::LpFrameHeader},
};
use nym_sphinx_addressing::nodes::NymNodeRoutingAddress;
use rand::Rng;
use tracing::warn;

use crate::node::{
    lp::data::{
        handler::{
            messages::MixMessage,
            pipeline::{NymNodeDataPipeline, wire::WirePipeline},
        },
        shared::SharedLpDataState,
    },
    routing_filter::RoutingFilter,
};

pub(crate) struct MixingNodeDataPipeline<R> {
    state: Arc<SharedLpDataState>,
    wire: WirePipeline<R>,
}

impl<R: Rng> MixingNodeDataPipeline<R> {
    pub(crate) fn new(state: Arc<SharedLpDataState>, rng: R) -> Self {
        Self {
            state: state.clone(),
            wire: WirePipeline::new(state, rng),
        }
    }
}

// Processing logic - stubbed; to be implemented.
impl<R: Rng> NymNodeProcessingPipeline<EncryptedLpPacket> for MixingNodeDataPipeline<R> {
    type Options = MixMessage;
    type MessageKind = MixMessage;

    fn mix(
        &mut self,
        message_kind: MixMessage,
        payload: TimedPayload,
        _: Instant,
    ) -> Vec<PipelinePayload<MixMessage>> {
        // Everything specific to a given packet type should happen here
        let processing_result =
            NymNodeDataPipeline::<R>::process_mix_packet(&self.state, message_kind, payload);

        self.state.update_processing_metrics(&processing_result);

        let packet_to_forward = match processing_result {
            Ok(packet) => packet,
            Err(e) => {
                warn!("Error processing {message_kind:?} packet : {e}");
                return Vec::new();
            }
        };

        // Now we are deciding if we are routing the packet and where

        match packet_to_forward.dst {
            NymNodeRoutingAddress::Node(next_hop) => {
                if !self.state.routing_filter.should_route(next_hop.ip()) {
                    warn!(
                        event = "packet.dropped.routing_filter",
                        next_hop = %next_hop,
                        "dropping packet: egress address does not belong to any known node"
                    );
                    self.state.routing_filter_dropped(next_hop);
                    Vec::new()
                } else {
                    vec![packet_to_forward.with_dst(next_hop)]
                }
            }
            NymNodeRoutingAddress::Client(_) => {
                warn!(
                    event = "packet.dropped.client_forwarding_disabled",
                    "dropping packet destined to a client_address on a client_forwarding_disabled node"
                );
                self.state.client_forwarding_disabled_dropped();
                Vec::new()
            }
        }
    }
}

// ============== Wire wrap/unwrap: pure delegation to WirePipeline ==============

impl<R: Rng> Framing<MixMessage> for MixingNodeDataPipeline<R> {
    type Frame = LpFrame;
    const OVERHEAD_SIZE: usize = LpFrameHeader::SIZE;

    fn to_frame(
        &mut self,
        payload: PipelinePayload<MixMessage>,
        frame_size: usize,
    ) -> Vec<AddressedTimedData<Self::Frame>> {
        let frame = LpFrame {
            header: payload.options.into(),
            content: payload.data.data.into(),
        };
        self.wire
            .message_to_frame(payload.data.timestamp, frame, payload.dst, frame_size)
    }
}

impl<R: Rng> Transport<EncryptedLpPacket> for MixingNodeDataPipeline<R> {
    type Frame = LpFrame;
    const OVERHEAD_SIZE: usize = LpHeader::SIZE;

    fn to_transport_packet(
        &mut self,
        frame: AddressedTimedData<Self::Frame>,
    ) -> AddressedTimedData<EncryptedLpPacket> {
        self.wire.frame_to_packet(frame)
    }
}

impl<R: Rng> WireWrappingPipeline<EncryptedLpPacket, MixMessage> for MixingNodeDataPipeline<R> {
    fn packet_size(&self) -> usize {
        nym_lp_data::packet::MTU
    }
}

impl<R: Rng> TransportUnwrap<EncryptedLpPacket> for MixingNodeDataPipeline<R> {
    type Frame = LpFrame;
    type Error = MalformedLpPacketError;

    fn packet_to_frame(
        &mut self,
        packet: EncryptedLpPacket,
        timestamp: Instant,
    ) -> Result<TimedData<Self::Frame>, Self::Error> {
        self.wire.packet_to_frame(packet, timestamp)
    }
}

impl<R: Rng> FramingUnwrap<MixMessage> for MixingNodeDataPipeline<R> {
    type Frame = LpFrame;

    fn frame_to_message(
        &mut self,
        frame: TimedData<Self::Frame>,
    ) -> Option<(TimedPayload, MixMessage)> {
        let reassembled = self.wire.frame_to_maybe_message(frame)?;
        let message_kind = MixMessage::from_frame_header(reassembled.data.header)
            .inspect_err(|e| warn!("{e}"))
            .ok()?;

        self.state.message_received(message_kind);
        Some((
            TimedPayload::new(reassembled.timestamp, reassembled.data.content.to_vec()),
            message_kind,
        ))
    }
}

impl<R: Rng> WireUnwrappingPipeline<EncryptedLpPacket, MixMessage> for MixingNodeDataPipeline<R> {}