linkrs-api 0.1.0

//! C API Session Management Module
//!
//! It provides functions for creating, destroying, and performing basic management of sessions.

use crate::embedded::c_api::database::LinkrsHandle;
use crate::embedded::c_api::error::{
    error_code_from_core_error, extended_error_code_from_core_error, linkrs_error_code_t,
    set_last_error_message,
};
use crate::embedded::c_api::types::{
    linkrs_commit_hook_callback, linkrs_extended_error_code_t, linkrs_rollback_hook_callback,
    linkrs_session_t, linkrs_t, linkrs_trace_callback, linkrs_update_hook_callback,
};
use crate::embedded::Session;
use linkrs_storage::GraphStorage;
use std::ffi::{c_char, c_int, c_void, CStr, CString};
use std::ptr;

/// Internal structure of the session handler
pub struct LinkrsSessionHandle {
    pub(crate) inner: Session<GraphStorage>,
    pub(crate) last_error: Option<CString>,
    /// Busy wait timeout (in milliseconds)
    pub(crate) busy_timeout_ms: u32,
    /// Offset of the last error location
    pub(crate) last_error_offset: Option<usize>,
    /// Finally, the error code was expanded.
    pub(crate) last_extended_error: Option<linkrs_extended_error_code_t>,
    /// SQL tracing callback
    pub(crate) trace_callback: linkrs_trace_callback,
    /// SQL tracking callback for user data
    pub(crate) trace_user_data: *mut c_void,
    /// User data for the commit hook, returned to the caller on replacement.
    pub(crate) commit_hook_user_data: *mut c_void,
    /// Bridge registration id of the commit hook on the transaction
    /// manager's veto registry (`None` when no hook is installed). The C
    /// callback and the Rust veto share one evaluation point.
    pub(crate) commit_veto_id: Option<linkrs_core::event_dispatch::SubscriptionId>,
    /// Rollback hook callback
    pub(crate) rollback_hook: linkrs_rollback_hook_callback,
    /// Roll back the user data associated with the hook.
    pub(crate) rollback_hook_user_data: *mut c_void,
    /// Update the hook callback
    pub(crate) update_hook: linkrs_update_hook_callback,
    /// Update the user data for the hook.
    pub(crate) update_hook_user_data: *mut c_void,
}

// The Send and Sync functions need to be implemented manually, because the type *mut c_void is not thread-safe.
// Thread-safety contract: a session handle must not be shared across threads
// concurrently. Hook callbacks (commit veto, update, trace) always run on the
// thread performing the operation that triggers them; `user_data` must stay
// valid from registration until replacement, session close, or explicit NULL
// deregistration, whichever comes first.
unsafe impl Send for LinkrsSessionHandle {}
unsafe impl Sync for LinkrsSessionHandle {}

impl LinkrsSessionHandle {
    /// Create a new session handle.
    pub(crate) fn new(inner: Session<GraphStorage>) -> Self {
        Self {
            inner,
            last_error: None,
            busy_timeout_ms: 5000, // Default value: 5 seconds
            last_error_offset: None,
            last_extended_error: None,
            trace_callback: None,
            trace_user_data: ptr::null_mut(),
            commit_hook_user_data: ptr::null_mut(),
            commit_veto_id: None,
            rollback_hook: None,
            rollback_hook_user_data: ptr::null_mut(),
            update_hook: None,
            update_hook_user_data: ptr::null_mut(),
        }
    }

    /// Whether an update hook is installed.
    ///
    /// Fast path so query execution can skip statement classification
    /// (`detect_data_modification`, including a full-string uppercase scan
    /// for `MATCH`-family queries) when nobody listens.
    pub(crate) fn has_update_hook(&self) -> bool {
        self.update_hook.is_some()
    }

    /// Call the update hook.
    ///
    /// Statement-level DML notification (not per-row): `database` carries
    /// the space name and `table` is an empty string (there is no table
    /// concept). `rowid` carries the affected row count from the result
    /// metadata, not a stable row id.
    ///
    /// The callback runs with panic isolation: a panicking C callback is
    /// caught, logged, and skipped so the query result is still returned.
    /// C callbacks must not unwind across the FFI boundary.
    pub(crate) fn invoke_update_hook(&self, operation: i32, space_name: &str, rowid: i64) {
        if let Some(callback) = self.update_hook {
            if let Ok(c_space) = CString::new(space_name) {
                // For graph databases, the `table` parameter should be set to an empty string.
                if let Ok(empty_table) = CString::new("") {
                    let user_data = self.update_hook_user_data;
                    let space_ptr = c_space.as_ptr();
                    let table_ptr = empty_table.as_ptr();
                    if std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
                        callback(user_data, operation, space_ptr, table_ptr, rowid);
                    }))
                    .is_err()
                    {
                        log::error!("update hook panicked; continuing query");
                    }
                }
            }
        }
    }

    /// Calling the SQL tracing callback with panic isolation.
    pub(crate) fn trace(&self, sql: &str) {
        if let Some(callback) = self.trace_callback {
            if let Ok(c_sql) = CString::new(sql) {
                let user_data = self.trace_user_data;
                let sql_ptr = c_sql.as_ptr();
                if std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
                    callback(sql_ptr, user_data);
                }))
                .is_err()
                {
                    log::error!("trace callback panicked; continuing query");
                }
            }
        }
    }

    /// Setting error messages
    pub(crate) fn set_error(
        &mut self,
        message: String,
        offset: Option<usize>,
        extended_code: Option<linkrs_extended_error_code_t>,
    ) {
        self.last_error = CString::new(message.clone()).ok();
        self.last_error_offset = offset;
        self.last_extended_error = extended_code;
        set_last_error_message(message);
    }

    /// Clear the error messages.
    pub(crate) fn clear_error(&mut self) {
        self.last_error = None;
        self.last_error_offset = None;
        self.last_extended_error = None;
    }
}

/// Create a session
///
/// # Arguments
/// - `db`: Database handle
/// - `session`: Output parameter, session handle
///
/// # Returns
/// - Success: LINKRS_OK
/// - Failure: Error code
///
/// # Safety
/// - `db` must be a valid database handle created by `linkrs_open` or `linkrs_open_with_config`
/// - `session` must be a valid pointer to store the session handle
/// - The caller is responsible for closing the session using `linkrs_session_close` when done
/// - The session handle must not be used after closing
#[no_mangle]
pub unsafe extern "C" fn linkrs_session_create(
    db: *mut linkrs_t,
    session: *mut *mut linkrs_session_t,
) -> c_int {
    if db.is_null() || session.is_null() {
        return linkrs_error_code_t::LINKRS_MISUSE as c_int;
    }

    let db_handle = &*(db as *mut LinkrsHandle);

    match db_handle.inner.session() {
        Ok(sess) => {
            let mut handle = Box::new(LinkrsSessionHandle::new(sess));
            // Inherit the database default timeout so the C
            // `linkrs_config_set_timeout` setting takes effect instead of
            // the fixed 5s fallback inside the handle constructor.
            let timeout_ms = db_handle.inner.config().default_timeout.as_millis();
            handle.busy_timeout_ms = timeout_ms.min(u32::MAX as u128) as u32;
            *session = Box::into_raw(handle) as *mut linkrs_session_t;
            linkrs_error_code_t::LINKRS_OK as c_int
        }
        Err(e) => {
            let (error_code, _) = error_code_from_core_error(&e);
            let error_msg = format!("{}", e);
            set_last_error_message(error_msg);
            *session = ptr::null_mut();
            error_code
        }
    }
}

/// Close the session.
///
/// # Arguments
/// - `session`: Session handle
///
/// # Returns
/// - Success: LINKRS_OK
/// - Failure: Error code
///
/// # Safety
/// - `session` must be a valid session handle created by `linkrs_session_create`
/// - After calling this function, the session handle becomes invalid and must not be used
/// - All pending transactions will be rolled back
#[no_mangle]
pub unsafe extern "C" fn linkrs_session_close(session: *mut linkrs_session_t) -> c_int {
    if session.is_null() {
        return linkrs_error_code_t::LINKRS_MISUSE as c_int;
    }

    let handle = Box::from_raw(session as *mut LinkrsSessionHandle);
    // Retire the commit-hook bridge so a closed session's user_data can
    // never be invoked by a later commit on the shared manager.
    if let Some(id) = handle.commit_veto_id {
        handle.inner.txn_manager().unregister_commit_veto(id);
    }
    drop(handle);

    linkrs_error_code_t::LINKRS_OK as c_int
}

/// Switch to the image space
///
/// # Arguments
/// - `session`: Session handle
/// - `space_name`: Graph space name (UTF-8 encoded)
///
/// # Returns
/// - Success: LINKRS_OK
/// - Failure: Error code
///
/// # Safety
/// - `session` must be a valid session handle created by `linkrs_session_create`
/// - `space_name` must be a valid pointer to a null-terminated UTF-8 string
#[no_mangle]
pub unsafe extern "C" fn linkrs_session_use_space(
    session: *mut linkrs_session_t,
    space_name: *const c_char,
) -> c_int {
    if session.is_null() || space_name.is_null() {
        return linkrs_error_code_t::LINKRS_MISUSE as c_int;
    }

    let name_str = match CStr::from_ptr(space_name).to_str() {
        Ok(s) => s,
        Err(_) => return linkrs_error_code_t::LINKRS_MISUSE as c_int,
    };

    let handle = &mut *(session as *mut LinkrsSessionHandle);

    match handle.inner.use_space(name_str) {
        Ok(_) => {
            handle.clear_error();
            linkrs_error_code_t::LINKRS_OK as c_int
        }
        Err(e) => {
            let (error_code, _) = error_code_from_core_error(&e);
            let error_msg = format!("{}", e);
            let offset = e.error_offset();
            let extended_code = Some(extended_error_code_from_core_error(&e));
            handle.set_error(error_msg, offset, extended_code);
            error_code
        }
    }
}

/// Obtain the current image space
///
/// # Arguments
/// - `session`: Session handle
///
/// # Returns
/// - Current graph space name (UTF-8 encoded), returns NULL if none
///
/// # Memory Management
/// The returned string is dynamically allocated and must be freed by the caller using `linkrs_free_string`
/// to avoid memory leaks.
///
/// # Example
/// ```c
/// char* space = linkrs_session_current_space(session);
/// if (space) {
///     printf("Current space: %s\n", space);
///     linkrs_free_string(space);  // Must free
/// }
/// ```
///
/// # Safety
/// - `session` must be a valid session handle created by `linkrs_session_create`
/// - The returned pointer must be freed by the caller to avoid memory leaks
#[no_mangle]
pub unsafe extern "C" fn linkrs_session_current_space(
    session: *mut linkrs_session_t,
) -> *mut c_char {
    if session.is_null() {
        return ptr::null_mut();
    }

    let handle = &*(session as *mut LinkrsSessionHandle);

    match handle.inner.current_space() {
        Some(name) => {
            match CString::new(name) {
                Ok(c_name) => {
                    // Convert a CString to the original pointer
                    // The caller needs to use `linkrs_free_string` to release the resource.
                    c_name.into_raw()
                }
                Err(_) => ptr::null_mut(),
            }
        }
        None => ptr::null_mut(),
    }
}

/// Enable the automatic submission mode.
///
/// # Arguments
/// - `session`: Session handle
/// - `autocommit`: Whether to enable autocommit
///
/// # Returns
/// - Success: LINKRS_OK
/// - Failure: Error code
///
/// # Safety
/// - `session` must be a valid session handle created by `linkrs_session_create`
#[no_mangle]
pub unsafe extern "C" fn linkrs_session_set_autocommit(
    session: *mut linkrs_session_t,
    autocommit: bool,
) -> c_int {
    if session.is_null() {
        return linkrs_error_code_t::LINKRS_MISUSE as c_int;
    }

    let handle = &mut *(session as *mut LinkrsSessionHandle);
    handle.inner.set_auto_commit(autocommit);
    linkrs_error_code_t::LINKRS_OK as c_int
}

/// Enable the automatic submission mode.
///
/// # Arguments
/// - `session`: Session handle
///
/// # Returns
/// - Whether autocommit is enabled
///
/// # Safety
/// - `session` must be a valid session handle created by `linkrs_session_create`
#[no_mangle]
pub unsafe extern "C" fn linkrs_session_get_autocommit(session: *mut linkrs_session_t) -> bool {
    if session.is_null() {
        return true; // Default automatic submission
    }

    let handle = &*(session as *mut LinkrsSessionHandle);
    handle.inner.auto_commit()
}

/// Setting the busy wait timeout
///
/// # Arguments
/// - `session`: Session handle
/// - `timeout_ms`: Timeout in milliseconds
///
/// # Returns
/// - Success: LINKRS_OK
/// - Failure: Error code
///
/// # Safety
/// - `session` must be a valid session handle created by `linkrs_session_create`
#[no_mangle]
pub unsafe extern "C" fn linkrs_busy_timeout(
    session: *mut linkrs_session_t,
    timeout_ms: c_int,
) -> c_int {
    if session.is_null() {
        return linkrs_error_code_t::LINKRS_MISUSE as c_int;
    }

    let handle = &mut *(session as *mut LinkrsSessionHandle);
    // The storage timeout settings are applied to the handle.
    handle.busy_timeout_ms = timeout_ms.max(0) as u32;
    linkrs_error_code_t::LINKRS_OK as c_int
}

/// Busy wait timeout has occurred.
///
/// # Arguments
/// - `session`: Session handle
///
/// # Returns
/// - Timeout in milliseconds, returns -1 on error
///
/// # Safety
/// - `session` must be a valid session handle created by `linkrs_session_create`
#[no_mangle]
pub unsafe extern "C" fn linkrs_busy_timeout_get(session: *mut linkrs_session_t) -> c_int {
    if session.is_null() {
        return 0;
    }

    let handle = &*(session as *mut LinkrsSessionHandle);
    handle.busy_timeout_ms as c_int
}

/// Setting up an SQL tracing callback
///
/// # Arguments
/// - `session`: Session handle
/// - `callback`: Trace callback function, NULL to disable tracing
/// - `user_data`: User data pointer, will be passed to the callback
///
/// # Returns
/// - Success: LINKRS_OK
/// - Failure: Error code
///
/// # Example
/// ```c
/// extern void my_trace_callback(const char* sql, void* data) {
///     printf("Executing: %s\n", sql);
/// }
///
/// linkrs_trace(session, my_trace_callback, NULL);
/// ```
///
/// # Safety
/// - `session` must be a valid session handle created by `linkrs_session_create`
/// - `callback` must be a valid function pointer, or NULL to disable tracing
/// - `user_data` is passed to the callback and must remain valid for the lifetime of the callback
#[no_mangle]
pub unsafe extern "C" fn linkrs_trace(
    session: *mut linkrs_session_t,
    callback: linkrs_trace_callback,
    user_data: *mut c_void,
) -> c_int {
    if session.is_null() {
        return linkrs_error_code_t::LINKRS_MISUSE as c_int;
    }

    let handle = &mut *(session as *mut LinkrsSessionHandle);
    handle.trace_callback = callback;
    handle.trace_user_data = user_data;
    linkrs_error_code_t::LINKRS_OK as c_int
}

/// Setting up the commit hook
///
/// # Arguments
/// - `session`: Session handle
/// - `callback`: Commit hook callback function, NULL to disable the hook
/// - `user_data`: User data pointer, will be passed to the callback
///
/// # Returns
/// - Previous hook user data pointer (if any)
///
/// # Description
/// The commit hook is called before a transaction is committed. If the callback returns a non-zero value,
/// the transaction will be rolled back.
///
/// The C callback is bridged into the transaction manager's commit-veto
/// registry, so it shares one evaluation point with Rust
/// `register_commit_veto` hooks (first veto wins). A panicking callback is
/// isolated and treated as allow.
///
/// # Safety
/// - `session` must be a valid session handle created by `linkrs_session_create`
/// - `callback` must be a valid function pointer, or NULL to disable the hook
/// - `user_data` is passed to the callback and must remain valid for the lifetime of the callback
#[no_mangle]
pub unsafe extern "C" fn linkrs_commit_hook(
    session: *mut linkrs_session_t,
    callback: linkrs_commit_hook_callback,
    user_data: *mut c_void,
) -> *mut c_void {
    if session.is_null() {
        return ptr::null_mut();
    }

    let handle = &mut *(session as *mut LinkrsSessionHandle);
    let old_user_data = handle.commit_hook_user_data;
    // Retire the previous bridge registration, if any.
    if let Some(id) = handle.commit_veto_id.take() {
        handle.inner.txn_manager().unregister_commit_veto(id);
    }
    handle.commit_hook_user_data = user_data;
    if let Some(func) = callback {
        // `usize` round-trips the raw pointer into the `'static` closure
        // without claiming `Send` for it; the callback always runs on the
        // thread performing the commit, and lifetime is bounded by the
        // unregister above / session close.
        let user_data_addr = user_data as usize;
        let bridge: std::sync::Arc<
            dyn Fn(&linkrs_transaction::CommitVetoContext) -> linkrs_transaction::VetoDecision
                + Send
                + Sync,
        > = std::sync::Arc::new(move |_| {
            let restored = user_data_addr as *mut c_void;
            let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| func(restored)));
            match outcome {
                Ok(0) => linkrs_transaction::VetoDecision::allow(),
                Ok(_) => linkrs_transaction::VetoDecision::veto("c commit hook vetoed"),
                Err(_) => {
                    log::error!("commit hook panicked; treating as allow");
                    linkrs_transaction::VetoDecision::allow()
                }
            }
        });
        let id = handle.inner.txn_manager().register_commit_veto(bridge);
        handle.commit_veto_id = Some(id);
    }
    old_user_data
}

/// Setting up a rollback hook
///
/// # Arguments
/// - `session`: Session handle
/// - `callback`: Rollback hook callback function, NULL to disable the hook
/// - `user_data`: User data pointer, will be passed to the callback
///
/// # Returns
/// - Previous hook user data pointer (if any)
///
/// # Safety
/// - `session` must be a valid session handle created by `linkrs_session_create`
/// - `callback` must be a valid function pointer, or NULL to disable the hook
/// - `user_data` is passed to the callback and must remain valid for the lifetime of the callback
#[no_mangle]
pub unsafe extern "C" fn linkrs_rollback_hook(
    session: *mut linkrs_session_t,
    callback: linkrs_rollback_hook_callback,
    user_data: *mut c_void,
) -> *mut c_void {
    if session.is_null() {
        return ptr::null_mut();
    }

    let handle = &mut *(session as *mut LinkrsSessionHandle);
    let old_user_data = handle.rollback_hook_user_data;
    handle.rollback_hook = callback;
    handle.rollback_hook_user_data = user_data;
    old_user_data
}

/// Set up the update hook
///
/// When data in the database changes, the callback function is called
///
/// # Arguments
/// - `session`: Session handle
/// - `callback`: Update hook callback function, NULL to disable the hook
/// - `user_data`: User data pointer, will be passed to the callback
///
/// # Returns
/// - Previous hook user data pointer (if any)
///
/// # Callback Parameters
/// - `operation`: Operation type (1=INSERT, 2=UPDATE, 3=DELETE)
/// - `database`: Database/space name
/// - `table`: Table name (empty string for graph database)
/// - `rowid`: Affected row ID
///
/// # Safety
/// - `session` must be a valid session handle created by `linkrs_session_create`
/// - `callback` must be a valid function pointer, or NULL to disable the hook
/// - `user_data` is passed to the callback and must remain valid for the lifetime of the callback
#[no_mangle]
pub unsafe extern "C" fn linkrs_update_hook(
    session: *mut linkrs_session_t,
    callback: linkrs_update_hook_callback,
    user_data: *mut c_void,
) -> *mut c_void {
    if session.is_null() {
        return ptr::null_mut();
    }

    let handle = &mut *(session as *mut LinkrsSessionHandle);
    let old_user_data = handle.update_hook_user_data;
    handle.update_hook = callback;
    handle.update_hook_user_data = user_data;
    old_user_data
}

/// Interrupt a session's next query.
///
/// Cooperative cancellation at the entry gate: sets the session interrupt
/// flag so the next `execute` entry point fails fast with an interrupted
/// error. A query already running is not touched; stop those through the
/// executor kill path. Use `linkrs_connection_clear_interrupt` to resume
/// normal execution.
///
/// # Safety
/// - `session` must be a valid session handle created by `linkrs_session_create`
#[no_mangle]
pub unsafe extern "C" fn linkrs_connection_interrupt(session: *mut linkrs_session_t) -> c_int {
    if session.is_null() {
        return linkrs_error_code_t::LINKRS_MISUSE as c_int;
    }

    let handle = &*(session as *const LinkrsSessionHandle);
    handle.inner.interrupt();
    linkrs_error_code_t::LINKRS_OK as c_int
}

/// Clear a previously requested session interrupt.
///
/// # Safety
/// - `session` must be a valid session handle created by `linkrs_session_create`
#[no_mangle]
pub unsafe extern "C" fn linkrs_connection_clear_interrupt(
    session: *mut linkrs_session_t,
) -> c_int {
    if session.is_null() {
        return linkrs_error_code_t::LINKRS_MISUSE as c_int;
    }

    let handle = &*(session as *const LinkrsSessionHandle);
    handle.inner.clear_interrupt();
    linkrs_error_code_t::LINKRS_OK as c_int
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::embedded::c_api::database::{linkrs_close, linkrs_open};
    use std::sync::atomic::{AtomicU64, Ordering};

    static TEST_COUNTER: AtomicU64 = AtomicU64::new(0);

    fn create_test_db() -> *mut linkrs_t {
        let counter = TEST_COUNTER.fetch_add(1, Ordering::SeqCst);
        let temp_dir = std::env::temp_dir().join("linkrs_c_api_test");
        std::fs::create_dir_all(&temp_dir).ok();
        let db_path = temp_dir.join(format!(
            "test_session_{}_{}.db",
            std::process::id(),
            counter
        ));

        // Make sure the database file does not exist.
        if db_path.exists() {
            std::fs::remove_file(&db_path).ok();
            // Waiting for the file system to complete the deletion operation.
            std::thread::sleep(std::time::Duration::from_millis(50));
        }

        let path_cstring = CString::new(db_path.to_str().expect("Invalid path"))
            .expect("Failed to create CString");
        let mut db: *mut linkrs_t = ptr::null_mut();

        let rc = unsafe { linkrs_open(path_cstring.as_ptr(), &mut db) };
        if rc != linkrs_error_code_t::LINKRS_OK as c_int {
            panic!(
                "Failed to open database, error code: {}, path: {:?}",
                rc, db_path
            );
        }
        assert!(!db.is_null());

        db
    }

    #[test]
    fn test_session_create_close() {
        let db = create_test_db();
        let mut session: *mut linkrs_session_t = ptr::null_mut();

        // Create a session
        let rc = unsafe { linkrs_session_create(db, &mut session) };
        assert_eq!(rc, linkrs_error_code_t::LINKRS_OK as c_int);
        assert!(!session.is_null());

        // Close the session.
        let rc = unsafe { linkrs_session_close(session) };
        assert_eq!(rc, linkrs_error_code_t::LINKRS_OK as c_int);

        // Close the database.
        let rc = unsafe { linkrs_close(db) };
        assert_eq!(rc, linkrs_error_code_t::LINKRS_OK as c_int);
    }

    #[test]
    fn test_session_autocommit() {
        let db = create_test_db();
        let mut session: *mut linkrs_session_t = ptr::null_mut();

        let rc = unsafe { linkrs_session_create(db, &mut session) };
        assert_eq!(rc, linkrs_error_code_t::LINKRS_OK as c_int);

        // Default automatic submission
        let autocommit = unsafe { linkrs_session_get_autocommit(session) };
        assert!(autocommit);

        // Turn off automatic submission.
        let rc = unsafe { linkrs_session_set_autocommit(session, false) };
        assert_eq!(rc, linkrs_error_code_t::LINKRS_OK as c_int);

        let autocommit = unsafe { linkrs_session_get_autocommit(session) };
        assert!(!autocommit);

        unsafe { linkrs_session_close(session) };
        unsafe { linkrs_close(db) };
    }
}