nautilus_plugin/surfaces/

custom_data.rs

// -------------------------------------------------------------------------------------------------
//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
//  https://nautechsystems.io
//
//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
//  You may not use this file except in compliance with the License.
//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
//
//  Unless required by applicable law or agreed to in writing, software
//  distributed under the License is distributed on an "AS IS" BASIS,
//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//  See the License for the specific language governing permissions and
//  limitations under the License.
// -------------------------------------------------------------------------------------------------

//! Custom-data plug point.
//!
//! A plug-in registers one [`CustomDataVTable`] per concrete custom-data
//! type. The host calls through the vtable to decode JSON envelopes from
//! the wire and to encode/decode Arrow IPC batches for catalog persistence.

#![allow(unsafe_code)]

use std::marker::PhantomData;

use crate::{
    boundary::{BorrowedStr, OwnedBytes, PluginError, PluginErrorCode, PluginResult, Slice},
    panic::{guard, guard_infallible},
};

/// Opaque handle to a single custom-data value owned by the plug-in.
///
/// The host never deref's this pointer; it only passes it back to the vtable
/// through `clone`, `to_json`, `ts_event`, `ts_init`, `eq`, and `drop`.
#[repr(C)]
pub struct CustomDataHandle {
    _opaque: [u8; 0],
}

/// Borrowed view of a plug-in custom-data value during `on_data` dispatch.
///
/// The value stays owned by the host-side wrapper. Plug-in code can inspect
/// the type name and downcast to a concrete custom-data type that was declared
/// in the same cdylib manifest.
#[repr(C)]
#[derive(Clone, Copy)]
pub struct PluginCustomDataRef {
    type_name: BorrowedStr<'static>,
    vtable: *const CustomDataVTable,
    handle: *const CustomDataHandle,
}

impl PluginCustomDataRef {
    /// Constructs a borrowed custom-data reference from raw boundary parts.
    ///
    /// # Safety
    ///
    /// `handle` must be null or point at a live value allocated by `vtable`,
    /// and `type_name` must name the same concrete custom-data type for the
    /// duration of the call.
    #[must_use]
    pub unsafe fn from_raw_parts(
        type_name: BorrowedStr<'static>,
        vtable: *const CustomDataVTable,
        handle: *const CustomDataHandle,
    ) -> Self {
        Self {
            type_name,
            vtable,
            handle,
        }
    }

    /// Returns the canonical custom-data type name.
    #[must_use]
    pub fn type_name(&self) -> &str {
        // SAFETY: constructor requires process-lifetime valid UTF-8 storage.
        unsafe { self.type_name.as_str() }
    }

    /// Returns whether this value was allocated by the vtable for `T`.
    #[must_use]
    pub fn is<T>(&self) -> bool
    where
        T: PluginCustomData + PartialEq + Clone,
    {
        self.vtable == custom_data_vtable::<T>()
    }

    /// Returns the value as `T` when it was allocated by `T`'s vtable.
    #[must_use]
    pub fn downcast_ref<T>(&self) -> Option<&T>
    where
        T: PluginCustomData + PartialEq + Clone,
    {
        if self.handle.is_null() || !self.is::<T>() {
            return None;
        }

        // SAFETY: matching vtable proves the handle came from `T`'s
        // generated custom-data thunks and is live for this callback.
        Some(unsafe { &*self.handle.cast::<T>() })
    }
}

/// Function table for a single custom-data type.
///
/// One static vtable per concrete type, generated by the `nautilus_plugin!`
/// macro and pointed to from the corresponding
/// [`crate::manifest::CustomDataRegistration`] entry.
///
/// Slots are nullable at the ABI type level so the host can reject malformed
/// manifests with null callbacks before registering or invoking the vtable.
/// Macro-generated vtables fill every required slot.
#[repr(C)]
pub struct CustomDataVTable {
    /// Returns the canonical type name (e.g. `"MyTickType"`). Used for
    /// routing, persistence path layout, and JSON envelope tagging.
    pub type_name: Option<unsafe extern "C" fn() -> BorrowedStr<'static>>,

    /// Returns the Arrow schema for this type as an Arrow IPC byte stream.
    ///
    /// Plug-ins serialize their `arrow::datatypes::Schema` via
    /// `arrow::ipc::writer::StreamWriter` so host and plug-in stay agnostic
    /// of each other's Arrow crate version.
    pub schema_ipc: Option<unsafe extern "C" fn() -> PluginResult<OwnedBytes>>,

    /// Decodes a single value from its JSON payload (no envelope) into a new
    /// handle. The handle is owned by the host until passed to `drop`.
    pub from_json: Option<
        unsafe extern "C" fn(payload: BorrowedStr<'_>) -> PluginResult<*mut CustomDataHandle>,
    >,

    /// Encodes a batch of handles into an Arrow IPC byte stream.
    pub encode_batch: Option<
        unsafe extern "C" fn(
            handles: Slice<'_, *const CustomDataHandle>,
        ) -> PluginResult<OwnedBytes>,
    >,

    /// Decodes an Arrow IPC byte stream into a freshly allocated array of
    /// owned handles, packed as `*mut CustomDataHandle` elements inside the
    /// returned `OwnedBytes`. The host must pass every handle through
    /// `drop_handle` to release its inner value, then drop the `OwnedBytes`
    /// (or invoke its embedded `drop_fn` directly) so the array storage is
    /// freed with the original allocator layout. Do not call
    /// [`crate::boundary::drop_owned_bytes`] directly: the underlying
    /// allocation is a `Vec<*mut CustomDataHandle>`, not a `Vec<u8>`, and the
    /// vtable installs a matching `drop_fn` on the returned `OwnedBytes`.
    pub decode_batch: Option<
        unsafe extern "C" fn(
            ipc_bytes: Slice<'_, u8>,
            metadata: Slice<'_, MetadataEntry<'_>>,
        ) -> PluginResult<OwnedBytes>,
    >,

    /// Returns the event timestamp of the value behind the handle.
    pub ts_event: Option<unsafe extern "C" fn(handle: *const CustomDataHandle) -> u64>,

    /// Returns the init timestamp of the value behind the handle.
    pub ts_init: Option<unsafe extern "C" fn(handle: *const CustomDataHandle) -> u64>,

    /// Serializes a single value to a JSON payload (no envelope).
    pub to_json:
        Option<unsafe extern "C" fn(handle: *const CustomDataHandle) -> PluginResult<OwnedBytes>>,

    /// Clones the value behind the handle into a new owned handle.
    pub clone_handle:
        Option<unsafe extern "C" fn(handle: *const CustomDataHandle) -> *mut CustomDataHandle>,

    /// Drops the value behind the handle, freeing all of its resources.
    pub drop_handle: Option<unsafe extern "C" fn(handle: *mut CustomDataHandle)>,

    /// Tests two handles of the same type for value equality.
    pub eq_handles: Option<
        unsafe extern "C" fn(lhs: *const CustomDataHandle, rhs: *const CustomDataHandle) -> bool,
    >,
}

/// A single key/value pair in batch-decode metadata.
#[repr(C)]
#[derive(Clone, Copy)]
pub struct MetadataEntry<'a> {
    pub key: BorrowedStr<'a>,
    pub value: BorrowedStr<'a>,
}

/// Author-facing trait for a custom-data type contributed by a plug-in.
///
/// Plug-in authors implement this on their data struct; the
/// [`nautilus_plugin!`](crate::nautilus_plugin) macro generates the
/// `extern "C"` thunks that adapt this trait to a [`CustomDataVTable`].
///
/// All trait methods run inside [`crate::panic::guard`] in the generated
/// thunks, so a panic surfaces as a [`PluginError`] with code
/// [`PluginErrorCode::Panic`] instead of unwinding through the FFI.
pub trait PluginCustomData: 'static + Send + Sync + Sized {
    /// Canonical type name. Must be unique across a Nautilus deployment.
    const TYPE_NAME: &'static str;

    /// Returns the event timestamp as UNIX nanoseconds.
    fn ts_event(&self) -> u64;

    /// Returns the init timestamp as UNIX nanoseconds.
    fn ts_init(&self) -> u64;

    /// Serializes this value to a JSON payload (no envelope, payload only).
    fn to_json(&self) -> anyhow::Result<Vec<u8>>;

    /// Deserializes a value from a JSON payload (no envelope).
    fn from_json(payload: &[u8]) -> anyhow::Result<Self>;

    /// Returns the Arrow schema for this type as an Arrow IPC byte stream.
    fn schema_ipc() -> anyhow::Result<Vec<u8>>;

    /// Encodes a batch of values into an Arrow IPC byte stream.
    fn encode_batch(items: &[&Self]) -> anyhow::Result<Vec<u8>>;

    /// Decodes an Arrow IPC byte stream into a vector of values.
    fn decode_batch(ipc_bytes: &[u8], metadata: &[(String, String)]) -> anyhow::Result<Vec<Self>>;

    /// Tests two values of this type for equality. Default implementation
    /// requires `PartialEq`; override if a custom comparison is needed.
    fn equals(&self, other: &Self) -> bool
    where
        Self: PartialEq,
    {
        self == other
    }

    /// Clones the value into a new heap allocation.
    #[must_use]
    fn clone_value(&self) -> Self
    where
        Self: Clone,
    {
        Clone::clone(self)
    }
}

/// Returns a `&'static CustomDataVTable` for the given `PluginCustomData` type.
///
/// One static vtable per concrete `T`. The vtable lives in const-promoted
/// static storage attached to a `PhantomData<T>`-tagged generic struct, so
/// each monomorphisation gets its own table.
///
/// Naive `static FOO: OnceLock<CustomDataVTable> = OnceLock::new()` inside a
/// generic function is unsound here: when `FOO`'s type does not mention `T`,
/// Rust shares the static across every monomorphisation and the first
/// initializer wins, so every later type collapses onto the first type's
/// thunks.
#[must_use]
pub fn custom_data_vtable<T>() -> *const CustomDataVTable
where
    T: PluginCustomData + PartialEq + Clone,
{
    &VTableTag::<T>::VTABLE
}

struct VTableTag<T>(PhantomData<T>);

impl<T> VTableTag<T>
where
    T: PluginCustomData + PartialEq + Clone,
{
    const VTABLE: CustomDataVTable = CustomDataVTable {
        type_name: Some(type_name_thunk::<T>),
        schema_ipc: Some(schema_ipc_thunk::<T>),
        from_json: Some(from_json_thunk::<T>),
        encode_batch: Some(encode_batch_thunk::<T>),
        decode_batch: Some(decode_batch_thunk::<T>),
        ts_event: Some(ts_event_thunk::<T>),
        ts_init: Some(ts_init_thunk::<T>),
        to_json: Some(to_json_thunk::<T>),
        clone_handle: Some(clone_handle_thunk::<T>),
        drop_handle: Some(drop_handle_thunk::<T>),
        eq_handles: Some(eq_handles_thunk::<T>),
    };
}

unsafe extern "C" fn type_name_thunk<T: PluginCustomData>() -> BorrowedStr<'static> {
    BorrowedStr::from_str(T::TYPE_NAME)
}

unsafe extern "C" fn schema_ipc_thunk<T: PluginCustomData>() -> PluginResult<OwnedBytes> {
    guard(|| {
        T::schema_ipc()
            .map(OwnedBytes::from_vec)
            .map_err(|e| PluginError::new(PluginErrorCode::SerializationFailed, e.to_string()))
    })
}

unsafe extern "C" fn from_json_thunk<T: PluginCustomData>(
    payload: BorrowedStr<'_>,
) -> PluginResult<*mut CustomDataHandle> {
    guard(|| {
        // SAFETY: caller upholds liveness of `payload`'s storage across the call.
        let bytes = unsafe { payload.as_str() }.as_bytes();
        T::from_json(bytes)
            .map(|v| Box::into_raw(Box::new(v)).cast::<CustomDataHandle>())
            .map_err(|e| PluginError::new(PluginErrorCode::SerializationFailed, e.to_string()))
    })
}

unsafe extern "C" fn encode_batch_thunk<T: PluginCustomData>(
    handles: Slice<'_, *const CustomDataHandle>,
) -> PluginResult<OwnedBytes> {
    guard(|| {
        // SAFETY: caller upholds liveness of the handles slice and each handle.
        let raw = unsafe { handles.as_slice() };
        let items: Vec<&T> = raw
            .iter()
            .map(|p| {
                // SAFETY: each handle originated from a `Box::into_raw(Box::new(T))`
                // in `from_json_thunk` or `clone_handle_thunk`. The host promises
                // they are still live.
                unsafe { &*p.cast::<T>() }
            })
            .collect();
        T::encode_batch(&items)
            .map(OwnedBytes::from_vec)
            .map_err(|e| PluginError::new(PluginErrorCode::SerializationFailed, e.to_string()))
    })
}

unsafe extern "C" fn decode_batch_thunk<T: PluginCustomData>(
    ipc_bytes: Slice<'_, u8>,
    metadata: Slice<'_, MetadataEntry<'_>>,
) -> PluginResult<OwnedBytes> {
    guard(|| {
        // SAFETY: caller upholds liveness of `ipc_bytes` and `metadata`.
        let bytes = unsafe { ipc_bytes.as_slice() };
        // SAFETY: see above.
        let entries = unsafe { metadata.as_slice() };
        let md: Vec<(String, String)> = entries
            .iter()
            .map(|e| {
                // SAFETY: caller upholds storage liveness for key.
                let k = unsafe { e.key.as_str() }.to_string();
                // SAFETY: caller upholds storage liveness for value.
                let v = unsafe { e.value.as_str() }.to_string();
                (k, v)
            })
            .collect();
        let values: Vec<T> = T::decode_batch(bytes, &md)
            .map_err(|e| PluginError::new(PluginErrorCode::SerializationFailed, e.to_string()))?;

        // Pack the decoded values into a contiguous run of opaque handles.
        // The host owns the buffer and must hand every handle back to
        // `drop_handle` before freeing the array via `drop_fn`.
        let handles: Vec<*mut CustomDataHandle> = values
            .into_iter()
            .map(|v| Box::into_raw(Box::new(v)).cast::<CustomDataHandle>())
            .collect();
        let mut handles = std::mem::ManuallyDrop::new(handles);
        let ptr = handles.as_mut_ptr().cast::<u8>();
        let elem_size = std::mem::size_of::<*mut CustomDataHandle>();
        let len = handles.len() * elem_size;
        let cap = handles.capacity() * elem_size;

        // `drop_fn` reconstructs the Vec with the original element layout so
        // the allocator deallocates with matching alignment; using
        // `drop_owned_bytes` here would call `dealloc` with the wrong layout
        // (Vec<u8> has 1-byte alignment, Vec<*mut _> has pointer alignment).
        Ok(OwnedBytes {
            ptr,
            len,
            cap,
            drop_fn: Some(drop_handle_array),
        })
    })
}

unsafe extern "C" fn drop_handle_array(ptr: *mut u8, len: usize, cap: usize) {
    if ptr.is_null() {
        return;
    }
    let elem_size = std::mem::size_of::<*mut CustomDataHandle>();
    if elem_size == 0 {
        return;
    }
    let len_count = len / elem_size;
    let cap_count = cap / elem_size;
    // SAFETY: ptr/len/cap originate from a `Vec<*mut CustomDataHandle>`
    // leaked via `ManuallyDrop` in `decode_batch_thunk`.
    unsafe {
        let _ = Vec::from_raw_parts(ptr.cast::<*mut CustomDataHandle>(), len_count, cap_count);
    }
}

unsafe extern "C" fn ts_event_thunk<T: PluginCustomData>(handle: *const CustomDataHandle) -> u64 {
    guard_infallible("ts_event", || {
        // SAFETY: handle originated from a `Box::into_raw(Box::new(T))` and
        // is still live (the host has not called drop_handle on it).
        let value = unsafe { &*handle.cast::<T>() };
        value.ts_event()
    })
}

unsafe extern "C" fn ts_init_thunk<T: PluginCustomData>(handle: *const CustomDataHandle) -> u64 {
    guard_infallible("ts_init", || {
        // SAFETY: see ts_event_thunk.
        let value = unsafe { &*handle.cast::<T>() };
        value.ts_init()
    })
}

unsafe extern "C" fn to_json_thunk<T: PluginCustomData>(
    handle: *const CustomDataHandle,
) -> PluginResult<OwnedBytes> {
    guard(|| {
        // SAFETY: see ts_event_thunk.
        let value = unsafe { &*handle.cast::<T>() };
        value
            .to_json()
            .map(OwnedBytes::from_vec)
            .map_err(|e| PluginError::new(PluginErrorCode::SerializationFailed, e.to_string()))
    })
}

unsafe extern "C" fn clone_handle_thunk<T: PluginCustomData + Clone>(
    handle: *const CustomDataHandle,
) -> *mut CustomDataHandle {
    guard_infallible("clone_handle", || {
        // SAFETY: see ts_event_thunk.
        let value = unsafe { &*handle.cast::<T>() };
        let cloned = value.clone_value();
        Box::into_raw(Box::new(cloned)).cast::<CustomDataHandle>()
    })
}

unsafe extern "C" fn drop_handle_thunk<T: PluginCustomData>(handle: *mut CustomDataHandle) {
    if handle.is_null() {
        return;
    }
    guard_infallible("drop_handle", || {
        // SAFETY: handle was allocated via `Box::into_raw(Box::new(T))`.
        unsafe {
            drop(Box::from_raw(handle.cast::<T>()));
        }
    });
}

unsafe extern "C" fn eq_handles_thunk<T: PluginCustomData + PartialEq>(
    lhs: *const CustomDataHandle,
    rhs: *const CustomDataHandle,
) -> bool {
    guard_infallible("eq_handles", || {
        // SAFETY: see ts_event_thunk.
        let lhs = unsafe { &*lhs.cast::<T>() };
        // SAFETY: see ts_event_thunk.
        let rhs = unsafe { &*rhs.cast::<T>() };
        lhs.equals(rhs)
    })
}