Function handle_subscribe

nodedb/src/control/server/sync/shape/handler.rs:97–132

Handle a ShapeSubscribe message. Registers the shape in the registry, evaluates the initial dataset via the provided `snapshot_provider`, and returns a ShapeSnapshot frame. The `snapshot_provider` callback bridges to the Data Plane: it receives the shape definition and current LSN, dispatches the appropriate query (DocumentScan for document shapes, GraphHop for graph shapes, collection scan for

(
    session_id: &str,
    tenant_id: u64,
    msg: &ShapeSubscribeMsg,
    registry: &ShapeRegistry,
    current_lsn: u64,
    snapshot_provider: F,
)

/// callback is synchronous because the SPSC bridge response is collected
/// before this function is called.
pub fn handle_subscribe<F>(
    session_id: &str,
    tenant_id: u64,
    msg: &ShapeSubscribeMsg,
    registry: &ShapeRegistry,
    current_lsn: u64,
    snapshot_provider: F,
) -> Option<SyncFrame>
where
    F: FnOnce(&ShapeDefinition, u64) -> ShapeSnapshotData,
{
    let shape = msg.shape.clone();
    let shape_id = shape.shape_id.clone();

    registry.subscribe(session_id, tenant_id, shape.clone());

    // Query the Data Plane for the initial dataset matching this shape.
    let snapshot_data = snapshot_provider(&shape, current_lsn);

    let snapshot = ShapeSnapshotMsg {
        shape_id,
        data: snapshot_data.data,
        snapshot_lsn: current_lsn,
        doc_count: snapshot_data.doc_count,
    };

    info!(
        session = session_id,
        shape_id = %msg.shape.shape_id,
        lsn = current_lsn,
        doc_count = snapshot.doc_count,
        "shape subscribed, snapshot sent"
    );

    SyncFrame::try_encode(SyncMessageType::ShapeSnapshot, &snapshot)
}

/// Handle a ShapeUnsubscribe message.
pub fn handle_unsubscribe(session_id: &str, msg: &ShapeUnsubscribeMsg, registry: &ShapeRegistry) {