linkrs_query/
dml_events.rs

1//! Statement-level DML classification and notification.
2//!
3//! Graph databases have no per-row identity to report, so the classic
4//! per-row update hook degrades to a statement-level notification: which
5//! kind of write a statement performed, in which space, affecting how many
6//! rows. [`classify_dml`] is the single shared classifier used by the
7//! embedded session path and the C-API update hook, so both report the
8//! same operation for the same text.
9//!
10//! The mapping is a best-effort heuristic over the leading keyword (after
11//! skipping whitespace and `--` / `//` / `#` / `/* */` comments); the
12//! affected-rows count always comes from the result metadata and must be
13//! treated as a row count, never as a stable row id.
14
15use std::sync::Arc;
16
17/// Kind of write a DML statement performed.
18#[derive(Debug, Clone, Copy, PartialEq, Eq)]
19pub enum DmlOp {
20    Insert,
21    Update,
22    Delete,
23}
24
25/// Statement-level DML notification.
26///
27/// Emitted after a data-modifying statement succeeds. Valid only for the
28/// dispatch call; observers must not retain it beyond the callback.
29#[derive(Debug, Clone)]
30pub struct DmlStatementEvent {
31    pub op: DmlOp,
32    pub space_name: String,
33    pub rows: u64,
34}
35
36/// Runtime observer for statement-level DML notifications.
37pub type DmlStatementCallback = Arc<dyn Fn(&DmlStatementEvent) + Send + Sync>;
38
39/// Classify a query text into a DML operation, or `None` for read-only
40/// (or unrecognized) statements.
41///
42/// `MATCH`/`WITH`/`UNWIND`/`OPTIONAL` openers carry the write in a trailing
43/// clause, so the whole text is scanned for a write keyword in that case.
44pub fn classify_dml(query_text: &str) -> Option<DmlOp> {
45    let keyword = leading_keyword(query_text);
46    if keyword.eq_ignore_ascii_case("MATCH")
47        || keyword.eq_ignore_ascii_case("WITH")
48        || keyword.eq_ignore_ascii_case("UNWIND")
49        || keyword.eq_ignore_ascii_case("OPTIONAL")
50    {
51        let upper = query_text.to_uppercase();
52        if upper.contains("DELETE") || upper.contains("DETACH DELETE") {
53            return Some(DmlOp::Delete);
54        }
55        if upper.contains("SET")
56            || upper.contains("REMOVE")
57            || upper.contains("MERGE")
58            || upper.contains("CREATE")
59        {
60            if upper.contains("MERGE") || upper.contains("CREATE") {
61                return Some(DmlOp::Insert);
62            }
63            return Some(DmlOp::Update);
64        }
65        return None;
66    }
67    if keyword.eq_ignore_ascii_case("INSERT")
68        || keyword.eq_ignore_ascii_case("CREATE")
69        || keyword.eq_ignore_ascii_case("MERGE")
70    {
71        return Some(DmlOp::Insert);
72    }
73    if keyword.eq_ignore_ascii_case("UPDATE")
74        || keyword.eq_ignore_ascii_case("SET")
75        || keyword.eq_ignore_ascii_case("REMOVE")
76        || keyword.eq_ignore_ascii_case("ALTER")
77    {
78        return Some(DmlOp::Update);
79    }
80    if keyword.eq_ignore_ascii_case("DELETE")
81        || keyword.eq_ignore_ascii_case("DETACH")
82        || keyword.eq_ignore_ascii_case("DROP")
83    {
84        return Some(DmlOp::Delete);
85    }
86    None
87}
88
89/// Extract the first keyword token, skipping whitespace and SQL/Cypher comments.
90fn leading_keyword(query: &str) -> &str {
91    let bytes = query.as_bytes();
92    let mut pos = 0;
93    while pos < bytes.len() {
94        while pos < bytes.len() && bytes[pos].is_ascii_whitespace() {
95            pos += 1;
96        }
97        if pos >= bytes.len() {
98            break;
99        }
100        if query[pos..].starts_with("--")
101            || query[pos..].starts_with("//")
102            || query[pos..].starts_with('#')
103        {
104            while pos < bytes.len() && bytes[pos] != b'\n' {
105                pos += 1;
106            }
107            continue;
108        }
109        if query[pos..].starts_with("/*") {
110            if let Some(end) = query[pos..].find("*/") {
111                pos += end + 2;
112                continue;
113            }
114            return "";
115        }
116        break;
117    }
118    let rest = &query[pos..];
119    let end = rest
120        .find(|c: char| !c.is_ascii_alphanumeric() && c != '_' && c != ':')
121        .unwrap_or(rest.len());
122    let mut token = &rest[..end];
123    // Handle `OPTIONAL MATCH` / `DETACH DELETE` two-word openers.
124    if token.eq_ignore_ascii_case("OPTIONAL") || token.eq_ignore_ascii_case("DETACH") {
125        let after: &str = rest[end..].trim_start();
126        let after_end = after
127            .find(|c: char| !c.is_ascii_alphanumeric() && c != '_' && c != ':')
128            .unwrap_or(after.len());
129        if !after[..after_end].is_empty() {
130            token = &after[..after_end];
131        }
132    }
133    token
134}
135
136#[cfg(test)]
137mod tests {
138    use super::*;
139
140    #[test]
141    fn leading_keywords_classify() {
142        assert_eq!(classify_dml("INSERT INTO t"), Some(DmlOp::Insert));
143        assert_eq!(classify_dml("  CREATE TAG u"), Some(DmlOp::Insert));
144        assert_eq!(classify_dml("-- comment\nMERGE (n)"), Some(DmlOp::Insert));
145        assert_eq!(classify_dml("UPDATE t SET a"), Some(DmlOp::Update));
146        assert_eq!(classify_dml("/* c */ SET n.a = 1"), Some(DmlOp::Update));
147        assert_eq!(classify_dml("DELETE FROM t"), Some(DmlOp::Delete));
148        assert_eq!(classify_dml("DETACH DELETE n"), Some(DmlOp::Delete));
149        assert_eq!(classify_dml("DROP TAG t"), Some(DmlOp::Delete));
150    }
151
152    #[test]
153    fn match_family_scans_trailing_clause() {
154        assert_eq!(classify_dml("MATCH (n) DELETE n"), Some(DmlOp::Delete));
155        assert_eq!(
156            classify_dml("MATCH (n) SET n.a = 1 RETURN n"),
157            Some(DmlOp::Update)
158        );
159        assert_eq!(
160            classify_dml("MATCH (n) RETURN n"),
161            None,
162            "pure reads stay silent"
163        );
164    }
165}
linkrs_query/dml_events.rs

linkrs_query/
dml_events.rs
