Function handle_sync_session

nodedb/src/control/server/sync/session_handler.rs:26–576

Handle one sync session with full RLS, audit, DLQ wired in.

(
    mut ws: tokio_tungstenite::WebSocketStream<tokio::net::TcpStream>,
    addr: SocketAddr,
    state: &SyncListenerState,
    shared: Option<Arc<SharedState>>,
)

/// Handle one sync session with full RLS, audit, DLQ wired in.
pub(super) async fn handle_sync_session(
    mut ws: tokio_tungstenite::WebSocketStream<tokio::net::TcpStream>,
    addr: SocketAddr,
    state: &SyncListenerState,
    shared: Option<Arc<SharedState>>,
) {
    use futures::{SinkExt, StreamExt};
    use tokio_tungstenite::tungstenite::Message;

    let session_id = format!(
        "sync-{addr}-{}",
        state.connections_accepted.load(Ordering::Relaxed)
    );
    let mut session =
        super::session::SyncSession::with_rate_limit(session_id.clone(), &state.config.rate_limit);
    session.device_metadata.remote_addr = addr.to_string();

    let jwt_validator =
        crate::control::security::jwt::JwtValidator::new(state.config.jwt_config.clone());

    let mut crdt_delivery_rx: Option<
        tokio::sync::mpsc::Receiver<crate::event::crdt_sync::types::OutboundDelta>,
    > = None;
    let mut crdt_control_rx: Option<
        tokio::sync::mpsc::Receiver<nodedb_types::sync::wire::SyncFrame>,
    > = None;
    let mut crdt_registered = false;

    let mut presence_rx: Option<tokio::sync::mpsc::Receiver<std::sync::Arc<Vec<u8>>>> = None;
    let mut presence_registered = false;

    let array_inbound: Option<Arc<crate::control::array_sync::OriginArrayInbound>> =
        shared.as_ref().map(|s| {
            let engine = Arc::new(crate::control::array_sync::OriginApplyEngine::new(
                Arc::clone(&s.array_sync_schemas),
                Arc::clone(&s.array_sync_op_log),
            ));
            let fanout = Arc::new(crate::control::array_sync::ArrayFanout::new(
                Arc::clone(&s.shape_registry),
                Arc::clone(&s.array_delivery),
                Arc::clone(&s.array_subscriber_cursors),
                Arc::clone(&s.array_snapshot_hlcs),
                Arc::clone(&s.array_merger_registry),
                0,
                0,
            ));
            let inbound = crate::control::array_sync::OriginArrayInbound::new(
                engine,
                Arc::clone(&s.array_sync_schemas),
                Arc::clone(s),
                crate::types::TenantId::new(0),
            )
            .with_observer(fanout);
            Arc::new(inbound)
        });

    let mut array_delivery_rx: Option<tokio::sync::mpsc::Receiver<Vec<u8>>> = None;
    let mut array_delivery_registered = false;