linkrs_server/graph_service/
cursor.rs

1use crate::client::cursor_context::CursorPage;
2use crate::storage::{
3    StorageClient, StorageOperationContextOps, StorageSchemaContextOps, StorageSyncContextOps,
4};
5
6use super::error::GraphServiceError;
7use super::GraphService;
8
9impl<
10        S: StorageClient
11            + StorageSchemaContextOps
12            + StorageSyncContextOps
13            + StorageOperationContextOps
14            + Clone
15            + 'static,
16    > GraphService<S>
17{
18    /// Open a forward-only cursor over a single statement's result.
19    pub async fn open_cursor(
20        &self,
21        session_id: i64,
22        stmt: &str,
23    ) -> Result<(u64, Vec<String>), String> {
24        let session = self
25            .session_manager
26            .find_session(session_id)
27            .ok_or_else(|| format!("Invalid session ID: {}", session_id))?;
28        if Self::is_command_like(stmt) {
29            return Err(
30                "Cursors support single data statements only; commands run on the materialized path"
31                    .to_string(),
32            );
33        }
34        if Self::has_multiple_statements(stmt) {
35            return Err(
36                "Cursors support single data statements only; run multi-statement scripts on the materialized path"
37                    .to_string(),
38            );
39        }
40        let result = self
41            .execute_stream(session_id, stmt)
42            .await
43            .map_err(|e: GraphServiceError| e.message().to_string())?;
44        let columns = result.column_names().unwrap_or_default();
45        let cursor_id = session.open_cursor(stmt.to_string(), result, columns.clone())?;
46        Ok((cursor_id, columns))
47    }
48
49    /// Fetch one page from a cursor. Chunk pulls run off the async
50    /// runtime; a failed execution drops the cursor and reports the error.
51    pub async fn fetch_cursor(
52        &self,
53        session_id: i64,
54        cursor_id: u64,
55        page_size: usize,
56    ) -> Result<CursorPage, String> {
57        let session = self
58            .session_manager
59            .find_session(session_id)
60            .ok_or_else(|| format!("Invalid session ID: {}", session_id))?;
61        tokio::task::spawn_blocking(move || session.fetch_cursor_page(cursor_id, page_size))
62            .await
63            .map_err(|e| format!("Cursor fetch task failed: {e}"))?
64    }
65
66    /// Release a cursor. Unknown ids report `closed: false`, never an error.
67    pub async fn close_cursor(&self, session_id: i64, cursor_id: u64) -> Result<bool, String> {
68        let session = self
69            .session_manager
70            .find_session(session_id)
71            .ok_or_else(|| format!("Invalid session ID: {}", session_id))?;
72        Ok(session.close_cursor(cursor_id))
73    }
74}
linkrs_server/graph_service/cursor.rs

linkrs_server/graph_service/
cursor.rs
