Skip to main content

kglite_cli/
lib.rs

1//! Shared implementation of the `kglite` CLI.
2//!
3//! `kglite` is an interactive Cypher shell for `.kgl` knowledge graphs, in the
4//! spirit of the `sqlite3` CLI: open a single file, run queries and
5//! dot-commands from the terminal, no Python or server required.
6//!
7//! Pure-Rust binary over `kglite::api::*` (no libpython link), mirroring the
8//! kglite-bolt-server / kglite-mcp-server crate pattern.
9
10mod exec;
11mod format;
12mod helper;
13mod migrate;
14mod repl;
15
16use std::collections::HashMap;
17use std::ffi::OsString;
18use std::io::{self, BufRead, Write};
19use std::path::{Path, PathBuf};
20use std::sync::Arc;
21use std::time::Duration;
22
23use anyhow::{Context, Result};
24use clap::{Parser, Subcommand, ValueEnum};
25use kglite::api::introspection::{
26    compute_description, ConnectionDetail, CypherDetail, FluentDetail,
27};
28use kglite::api::io::{
29    load_file, open_or_create_graph, save_graph, GraphFileIdentity, GraphWriterLease,
30    OpenDisposition,
31};
32use kglite::api::storage::{new_dir_graph_in_mode, StorageMode};
33use kglite::api::{DirGraph, Value};
34
35use crate::exec::QueryOptions;
36use crate::format::Mode;
37
38const WRITE_LOCK_TIMEOUT: Duration = Duration::from_secs(30);
39
40/// Interactive Cypher shell for kglite `.kgl` graphs.
41#[derive(Parser, Debug)]
42#[command(name = "kglite", version, about)]
43#[command(args_conflicts_with_subcommands = true)]
44struct Cli {
45    /// Path to a `.kgl` file to open. If omitted (or the file does not exist
46    /// yet), the shell starts with a fresh in-memory graph.
47    graph: Option<PathBuf>,
48    #[command(subcommand)]
49    command: Option<Command>,
50}
51
52#[derive(Subcommand, Debug)]
53enum Command {
54    /// Run a read-only Cypher query against a `.kgl` graph and print the result.
55    Query {
56        /// Path to the `.kgl` file.
57        graph: PathBuf,
58        /// Cypher query string.
59        query: String,
60        /// Output format.
61        #[arg(long, value_enum, default_value_t = OutputFormat::Table)]
62        format: OutputFormat,
63        /// Opt in to the parallel runtime for this query. A hint: operators
64        /// that cannot partition deterministically, and queries below the
65        /// engine's runtime size gate, still run sequentially.
66        #[arg(long)]
67        parallel: bool,
68    },
69    /// Run a write-capable Cypher statement against a `.kgl` graph.
70    Write {
71        /// Path to the `.kgl` file.
72        graph: PathBuf,
73        /// Cypher statement.
74        query: String,
75        /// Output format.
76        #[arg(long, value_enum, default_value_t = OutputFormat::Table)]
77        format: OutputFormat,
78        /// Persist the graph after a successful statement.
79        #[arg(long)]
80        save: bool,
81        /// Comma-separated node-type whitelist for CREATE/SET mutations.
82        #[arg(long)]
83        write_scope: Option<String>,
84        /// Git SHA to stamp on auto_timestamp types.
85        #[arg(long)]
86        git_sha: Option<String>,
87        /// Actor id to stamp on auto_timestamp types.
88        #[arg(long)]
89        modified_by: Option<String>,
90    },
91    /// Print the dependency frontier from `CALL ready_set(...)`.
92    ReadySet {
93        /// Path to the `.kgl` file.
94        graph: PathBuf,
95        /// Dependency relationship type.
96        #[arg(long, default_value = "DEPENDS_ON")]
97        relationship: String,
98        /// Done predicate over `n`, for example: `n.status = "done"`.
99        #[arg(long)]
100        done: String,
101        /// Optional node type to include in the frontier.
102        #[arg(long)]
103        node_type: Option<String>,
104        /// Output format.
105        #[arg(long, value_enum, default_value_t = OutputFormat::Table)]
106        format: OutputFormat,
107    },
108    /// Print the XML graph description used by agents for structure discovery.
109    Describe {
110        /// Path to the `.kgl` file.
111        graph: PathBuf,
112        /// Comma-separated node types for focused detail.
113        #[arg(long)]
114        types: Option<String>,
115        /// Search node types by name.
116        #[arg(long)]
117        type_search: Option<String>,
118        /// Include connection overview.
119        #[arg(long)]
120        connections: bool,
121        /// Comma-separated connection types for deep-dive detail.
122        #[arg(long)]
123        connection_types: Option<String>,
124        /// Include compact Cypher reference.
125        #[arg(long)]
126        cypher: bool,
127        /// Comma-separated Cypher topics for detailed docs.
128        #[arg(long)]
129        cypher_topics: Option<String>,
130        /// Include compact fluent API reference.
131        #[arg(long)]
132        fluent: bool,
133        /// Comma-separated fluent API topics for detailed docs.
134        #[arg(long)]
135        fluent_topics: Option<String>,
136        /// Max `(source_type, target_type)` pairs for connection deep-dives.
137        #[arg(long)]
138        max_pairs: Option<usize>,
139        /// Truncate long sample strings to this many characters.
140        #[arg(long, default_value_t = 40)]
141        sample_truncate: usize,
142    },
143    /// Keep one graph loaded and process JSONL requests on stdin.
144    Session {
145        /// Path to the `.kgl` file.
146        graph: PathBuf,
147        /// Default output format for query/write responses.
148        #[arg(long, value_enum, default_value_t = OutputFormat::Json)]
149        format: OutputFormat,
150        /// Save the graph when the session exits successfully.
151        #[arg(long)]
152        save_on_exit: bool,
153        /// Comma-separated node-type whitelist for write requests.
154        #[arg(long)]
155        write_scope: Option<String>,
156        /// Git SHA to stamp on auto_timestamp types for write requests.
157        #[arg(long)]
158        git_sha: Option<String>,
159        /// Actor id to stamp on auto_timestamp types for write requests.
160        #[arg(long)]
161        modified_by: Option<String>,
162    },
163    /// Print a deterministic, human-readable text projection of a `.kgl` to
164    /// stdout — the canonical form for a git `textconv` diff filter. Set up:
165    /// `git config diff.kglite.textconv "kglite export-text"` +
166    /// `echo '*.kgl diff=kglite' >> .gitattributes`.
167    ExportText {
168        /// Path to the `.kgl` file.
169        file: PathBuf,
170    },
171    /// Show what changed between two `.kgl` graphs — a structural delta over the
172    /// deterministic text projection: `-` lines dropped from A, `+` lines added
173    /// in B (a node/edge whose properties changed shows as a `-`/`+` pair).
174    Diff {
175        /// The "before" `.kgl`.
176        a: PathBuf,
177        /// The "after" `.kgl`.
178        b: PathBuf,
179    },
180    /// Export a `.kgl` as a SQLite-dialect SQL script — the no-lock-in exit.
181    /// Node types become tables, connection types become link tables. Ingest
182    /// with: `sqlite3 target.db < dump.sql`.
183    ExportSqlite {
184        /// Path to the `.kgl` file.
185        graph: PathBuf,
186        /// Where to write the SQL script. Omit to write to stdout.
187        output: Option<PathBuf>,
188    },
189    /// Apply pending Cypher migrations and advance the graph's user-schema
190    /// version. Migrations are `<version>_<name>.cypher` files in one
191    /// directory, applied in ascending version order; a re-run is a no-op.
192    Migrate {
193        /// Path to the `.kgl` file.
194        graph: PathBuf,
195        /// Directory holding the `<version>_<name>.cypher` migrations.
196        directory: PathBuf,
197        /// Print the plan without applying or saving anything.
198        #[arg(long)]
199        dry_run: bool,
200    },
201    /// Print the graph's user-schema version (the caller's own data-model
202    /// revision, distinct from the `.kgl` format version).
203    SchemaVersion {
204        /// Path to the `.kgl` file.
205        graph: PathBuf,
206        /// Stamp this version instead of printing — the "adopt migrations on an
207        /// existing graph" operation. Runs nothing; it asserts that the data
208        /// already has the shape migrations up to this version produce.
209        #[arg(long)]
210        set: Option<u32>,
211    },
212}
213
214#[derive(Clone, Copy, Debug, Default, ValueEnum)]
215enum OutputFormat {
216    #[default]
217    Table,
218    Csv,
219    Json,
220}
221
222impl From<OutputFormat> for Mode {
223    fn from(value: OutputFormat) -> Self {
224        match value {
225            OutputFormat::Table => Mode::Table,
226            OutputFormat::Csv => Mode::Csv,
227            OutputFormat::Json => Mode::Json,
228        }
229    }
230}
231
232fn open_text(path: &Path) -> Result<String> {
233    let p = path.to_string_lossy().to_string();
234    let g = load_file(&p).with_context(|| format!("failed to open {p}"))?;
235    Ok(kglite::api::io::to_text(&g))
236}
237
238/// Run the CLI over an explicit argument vector, including the program name.
239///
240/// The standalone binary and the `pip install kglite` wheel shim both call
241/// this entry point, so command parsing and behavior cannot drift.
242pub fn run<I, T>(args: I) -> Result<()>
243where
244    I: IntoIterator<Item = T>,
245    T: Into<OsString> + Clone,
246{
247    let cli = Cli::parse_from(args);
248
249    if let Some(Command::Query {
250        graph,
251        query,
252        format,
253        parallel,
254    }) = &cli.command
255    {
256        run_query(graph, query, (*format).into(), *parallel)?;
257        return Ok(());
258    }
259    if let Some(Command::Write {
260        graph,
261        query,
262        format,
263        save,
264        write_scope,
265        git_sha,
266        modified_by,
267    }) = &cli.command
268    {
269        run_write(
270            graph,
271            query,
272            (*format).into(),
273            *save,
274            write_scope.as_deref(),
275            git_sha.clone(),
276            modified_by.clone(),
277        )?;
278        return Ok(());
279    }
280    if let Some(Command::ReadySet {
281        graph,
282        relationship,
283        done,
284        node_type,
285        format,
286    }) = &cli.command
287    {
288        run_ready_set(
289            graph,
290            relationship,
291            done,
292            node_type.as_deref(),
293            (*format).into(),
294        )?;
295        return Ok(());
296    }
297    if let Some(Command::Describe {
298        graph,
299        types,
300        type_search,
301        connections,
302        connection_types,
303        cypher,
304        cypher_topics,
305        fluent,
306        fluent_topics,
307        max_pairs,
308        sample_truncate,
309    }) = &cli.command
310    {
311        run_describe(
312            graph,
313            DescribeOptions {
314                types: parse_csv(types.as_deref()),
315                type_search: type_search.clone(),
316                connections: detail_connections(*connections, connection_types.as_deref()),
317                cypher: detail_cypher(*cypher, cypher_topics.as_deref()),
318                fluent: detail_fluent(*fluent, fluent_topics.as_deref()),
319                max_pairs: *max_pairs,
320                sample_truncate: Some(*sample_truncate),
321            },
322        )?;
323        return Ok(());
324    }
325    if let Some(Command::Session {
326        graph,
327        format,
328        save_on_exit,
329        write_scope,
330        git_sha,
331        modified_by,
332    }) = &cli.command
333    {
334        run_session(
335            graph,
336            (*format).into(),
337            *save_on_exit,
338            write_scope.as_deref(),
339            git_sha.clone(),
340            modified_by.clone(),
341        )?;
342        return Ok(());
343    }
344    if let Some(Command::ExportText { file }) = &cli.command {
345        print!("{}", open_text(file)?);
346        return Ok(());
347    }
348    if let Some(Command::Diff { a, b }) = &cli.command {
349        let (ta, tb) = (open_text(a)?, open_text(b)?);
350        let a_lines: std::collections::BTreeSet<&str> =
351            ta.lines().filter(|l| !l.trim().is_empty()).collect();
352        let b_lines: std::collections::BTreeSet<&str> =
353            tb.lines().filter(|l| !l.trim().is_empty()).collect();
354        for l in a_lines.difference(&b_lines) {
355            println!("-{}", l.trim_start());
356        }
357        for l in b_lines.difference(&a_lines) {
358            println!("+{}", l.trim_start());
359        }
360        return Ok(());
361    }
362    if let Some(Command::ExportSqlite { graph, output }) = &cli.command {
363        return run_export_sqlite(graph, output.as_deref());
364    }
365    if let Some(Command::Migrate {
366        graph,
367        directory,
368        dry_run,
369    }) = &cli.command
370    {
371        return migrate::run(graph, directory, *dry_run);
372    }
373    if let Some(Command::SchemaVersion { graph, set }) = &cli.command {
374        return match set {
375            Some(version) => migrate::set_version(graph, *version),
376            None => migrate::print_version(graph),
377        };
378    }
379
380    let (graph, source, source_identity) = match &cli.graph {
381        Some(path) => {
382            let p = path.to_string_lossy().to_string();
383            let opened = open_or_create_graph(path, Some(StorageMode::Memory))
384                .with_context(|| format!("failed to open or create {}", path.display()))?;
385            if opened.disposition == OpenDisposition::Created {
386                eprintln!("note: {p} does not exist — starting an empty in-memory graph");
387            }
388            let source = (opened.disposition == OpenDisposition::Opened).then_some(p);
389            let identity = source.as_ref().map(|_| opened.identity);
390            (opened.graph, source, identity)
391        }
392        None => (Arc::new(fresh_graph()?), None, None),
393    };
394
395    repl::run(graph, source.as_deref(), source_identity)
396}
397
398/// A fresh in-memory graph. `new_dir_graph_in_mode` returns `Result<_, String>`
399/// (not an `Error`), so adapt it into `anyhow` explicitly.
400fn fresh_graph() -> Result<DirGraph> {
401    new_dir_graph_in_mode(StorageMode::Memory, None)
402        .map_err(|e| anyhow::anyhow!("failed to create an in-memory graph: {e}"))
403}
404
405fn load_graph(path: &Path) -> Result<Arc<DirGraph>> {
406    let p = path.to_string_lossy().to_string();
407    load_file(&p).with_context(|| format!("failed to open {p}"))
408}
409
410/// Write the SQL projection of a graph to a file, or to stdout when no output
411/// path is given — the same shape as `export-text`, so the dump can be piped
412/// straight into `sqlite3`.
413fn run_export_sqlite(graph_path: &Path, output: Option<&Path>) -> Result<()> {
414    let graph = load_graph(graph_path)?;
415    let sql = kglite::api::io::to_sqlite_dump(&graph, None)
416        .map_err(|e| anyhow::anyhow!("SQL export failed: {e}"))?;
417    match output {
418        Some(path) => {
419            std::fs::write(path, &sql)
420                .with_context(|| format!("failed to write {}", path.display()))?;
421            eprintln!("wrote {} ({} bytes)", path.display(), sql.len());
422        }
423        None => exec::write_stdout(&sql)?,
424    }
425    Ok(())
426}
427
428fn run_query(path: &Path, query: &str, mode: Mode, parallel: bool) -> Result<()> {
429    let graph = load_graph(path)?;
430    let (_, is_mutation) = kglite::api::cypher::parse_with_mutation_check(query)
431        .map_err(|e| anyhow::anyhow!("Cypher parse error: {e}"))?;
432    if is_mutation {
433        anyhow::bail!("query is read-only; use `kglite write` for mutations");
434    }
435    let params: HashMap<String, Value> = HashMap::new();
436    let options = QueryOptions {
437        parallel,
438        ..QueryOptions::default()
439    };
440    let outcome = exec::execute_readonly(&graph, query, &params, &options)
441        .with_context(|| "Cypher execution failed")?;
442    exec::write_stdout(&exec::render_outcome(mode, &outcome))?;
443    Ok(())
444}
445
446fn run_write(
447    path: &Path,
448    query: &str,
449    mode: Mode,
450    persist: bool,
451    write_scope: Option<&str>,
452    git_sha: Option<String>,
453    modified_by: Option<String>,
454) -> Result<()> {
455    let _lease = if persist {
456        Some(GraphWriterLease::acquire(path, WRITE_LOCK_TIMEOUT)?)
457    } else {
458        None
459    };
460    let mut graph = open_or_create_graph(path, persist.then_some(StorageMode::Memory))
461        .with_context(|| format!("failed to open or create {}", path.display()))?
462        .graph;
463    let params: HashMap<String, Value> = HashMap::new();
464    let options = QueryOptions {
465        write_scope: exec::parse_write_scope(write_scope),
466        git_sha,
467        modified_by,
468        ..QueryOptions::default()
469    };
470    let outcome = exec::execute(&mut graph, query, &params, &options)
471        .with_context(|| "Cypher execution failed")?;
472    if persist {
473        let p = path.to_string_lossy().to_string();
474        save_graph(&mut graph, &p).map_err(|e| anyhow::anyhow!("failed to save {p}: {e}"))?;
475    }
476    exec::write_stdout(&exec::render_outcome(mode, &outcome))?;
477    Ok(())
478}
479
480fn run_ready_set(
481    path: &Path,
482    relationship: &str,
483    done: &str,
484    node_type: Option<&str>,
485    mode: Mode,
486) -> Result<()> {
487    let mut config = vec![
488        format!("relationship: '{}'", cypher_string(relationship)),
489        format!("done: '{}'", cypher_string(done)),
490    ];
491    if let Some(node_type) = node_type {
492        config.push(format!("node_type: '{}'", cypher_string(node_type)));
493    }
494    let query = format!(
495        "CALL ready_set({{{}}}) YIELD node, dependency_count \
496         RETURN node.id AS id, node.title AS title, dependency_count \
497         ORDER BY dependency_count, id",
498        config.join(", ")
499    );
500    run_query(path, &query, mode, false)
501}
502
503struct DescribeOptions {
504    types: Option<Vec<String>>,
505    type_search: Option<String>,
506    connections: ConnectionDetail,
507    cypher: CypherDetail,
508    fluent: FluentDetail,
509    max_pairs: Option<usize>,
510    sample_truncate: Option<usize>,
511}
512
513fn run_describe(path: &Path, options: DescribeOptions) -> Result<()> {
514    let graph = load_graph(path)?;
515    let description = describe_graph(&graph, &options)?;
516    exec::write_stdout(&description)?;
517    Ok(())
518}
519
520fn describe_graph(graph: &Arc<DirGraph>, options: &DescribeOptions) -> Result<String> {
521    compute_description(
522        graph,
523        options.types.as_deref(),
524        &options.connections,
525        &options.cypher,
526        &options.fluent,
527        options.type_search.as_deref(),
528        options.max_pairs,
529        options.sample_truncate,
530    )
531    .map_err(|e| anyhow::anyhow!("describe failed: {e}"))
532}
533
534fn run_session(
535    path: &Path,
536    default_mode: Mode,
537    save_on_exit: bool,
538    write_scope: Option<&str>,
539    git_sha: Option<String>,
540    modified_by: Option<String>,
541) -> Result<()> {
542    let lease = if save_on_exit {
543        Some(GraphWriterLease::acquire(path, WRITE_LOCK_TIMEOUT)?)
544    } else {
545        None
546    };
547    let mut graph = open_or_create_graph(path, save_on_exit.then_some(StorageMode::Memory))
548        .with_context(|| format!("failed to open or create {}", path.display()))?
549        .graph;
550    let mut source_identity = GraphFileIdentity::capture(path)?;
551    let base_options = QueryOptions {
552        write_scope: exec::parse_write_scope(write_scope),
553        git_sha,
554        modified_by,
555        ..QueryOptions::default()
556    };
557    let stdin = io::stdin();
558    for line in stdin.lock().lines() {
559        let line = line?;
560        let line = line.trim();
561        if line.is_empty() {
562            continue;
563        }
564        match handle_session_line(
565            &mut graph,
566            path,
567            line,
568            default_mode,
569            &base_options,
570            &mut source_identity,
571            lease.is_some(),
572        ) {
573            SessionAction::Continue(value) => write_json_line(value)?,
574            SessionAction::Exit(value) => {
575                write_json_line(value)?;
576                if save_on_exit {
577                    save_loaded_graph(&mut graph, path, &mut source_identity, lease.is_some())?;
578                }
579                return Ok(());
580            }
581        }
582    }
583    if save_on_exit {
584        save_loaded_graph(&mut graph, path, &mut source_identity, lease.is_some())?;
585    }
586    Ok(())
587}
588
589enum SessionAction {
590    Continue(serde_json::Value),
591    Exit(serde_json::Value),
592}
593
594fn handle_session_line(
595    graph: &mut Arc<DirGraph>,
596    path: &Path,
597    line: &str,
598    default_mode: Mode,
599    base_options: &QueryOptions,
600    source_identity: &mut GraphFileIdentity,
601    lease_held: bool,
602) -> SessionAction {
603    let request: serde_json::Value = match serde_json::from_str(line) {
604        Ok(v) => v,
605        Err(e) => {
606            return SessionAction::Continue(json_error("parse", format!("invalid JSON: {e}")));
607        }
608    };
609    let op = request
610        .get("op")
611        .and_then(|v| v.as_str())
612        .unwrap_or("query");
613    let request_id = request.get("id").cloned();
614    let result = match op {
615        "query" => session_query(graph, &request, mode_from_request(&request, default_mode)),
616        "write" => session_write(
617            graph,
618            &request,
619            mode_from_request(&request, default_mode),
620            base_options,
621        ),
622        "describe" => session_describe(graph, &request),
623        "save" => save_loaded_graph(graph, path, source_identity, lease_held)
624            .map(|()| serde_json::json!({"ok": true, "op": "save"})),
625        "exit" | "quit" => {
626            let mut value = serde_json::json!({"ok": true, "op": op});
627            insert_request_id(&mut value, request_id);
628            return SessionAction::Exit(value);
629        }
630        other => Err(anyhow::anyhow!("unknown op {other:?}")),
631    };
632    SessionAction::Continue(match result {
633        Ok(mut value) => {
634            if let Some(obj) = value.as_object_mut() {
635                obj.entry("op").or_insert_with(|| serde_json::json!(op));
636            }
637            insert_request_id(&mut value, request_id);
638            value
639        }
640        Err(e) => {
641            let mut value = json_error(op, e.to_string());
642            insert_request_id(&mut value, request_id);
643            value
644        }
645    })
646}
647
648fn session_query(
649    graph: &Arc<DirGraph>,
650    request: &serde_json::Value,
651    mode: Mode,
652) -> Result<serde_json::Value> {
653    let query = request_string(request, "query")?;
654    let (_, is_mutation) = kglite::api::cypher::parse_with_mutation_check(&query)
655        .map_err(|e| anyhow::anyhow!("Cypher parse error: {e}"))?;
656    if is_mutation {
657        anyhow::bail!("query is read-only; use op=write for mutations");
658    }
659    let params = HashMap::new();
660    let outcome = exec::execute_readonly(graph, &query, &params, &QueryOptions::default())?;
661    Ok(session_outcome_response(mode, &outcome))
662}
663
664fn session_write(
665    graph: &mut Arc<DirGraph>,
666    request: &serde_json::Value,
667    mode: Mode,
668    base_options: &QueryOptions,
669) -> Result<serde_json::Value> {
670    let query = request_string(request, "query")?;
671    let params = HashMap::new();
672    let options = QueryOptions {
673        write_scope: request
674            .get("write_scope")
675            .and_then(json_string_vec)
676            .map(|v| v.into_iter().collect())
677            .or_else(|| base_options.write_scope.clone()),
678        git_sha: request
679            .get("git_sha")
680            .and_then(|v| v.as_str().map(str::to_string))
681            .or_else(|| base_options.git_sha.clone()),
682        modified_by: request
683            .get("modified_by")
684            .and_then(|v| v.as_str().map(str::to_string))
685            .or_else(|| base_options.modified_by.clone()),
686        ..QueryOptions::default()
687    };
688    let outcome = exec::execute(graph, &query, &params, &options)?;
689    Ok(session_outcome_response(mode, &outcome))
690}
691
692fn session_describe(
693    graph: &Arc<DirGraph>,
694    request: &serde_json::Value,
695) -> Result<serde_json::Value> {
696    let options = describe_options_from_json(request)?;
697    Ok(serde_json::json!({
698        "ok": true,
699        "description": describe_graph(graph, &options)?,
700    }))
701}
702
703fn save_loaded_graph(
704    graph: &mut Arc<DirGraph>,
705    path: &Path,
706    source_identity: &mut GraphFileIdentity,
707    lease_held: bool,
708) -> Result<()> {
709    let _lease = (!lease_held)
710        .then(|| GraphWriterLease::acquire(path, WRITE_LOCK_TIMEOUT))
711        .transpose()?;
712    let current = GraphFileIdentity::capture(path)?;
713    if current != *source_identity {
714        anyhow::bail!(
715            "refusing to overwrite {}: it changed since this session loaded it",
716            path.display()
717        );
718    }
719    let p = path.to_string_lossy().to_string();
720    save_graph(graph, &p).map_err(|e| anyhow::anyhow!("failed to save {p}: {e}"))?;
721    *source_identity = GraphFileIdentity::capture(path)?;
722    Ok(())
723}
724
725fn write_json_line(value: serde_json::Value) -> Result<()> {
726    let mut stdout = io::stdout().lock();
727    serde_json::to_writer(&mut stdout, &value)?;
728    stdout.write_all(b"\n")?;
729    stdout.flush()?;
730    Ok(())
731}
732
733fn session_outcome_response(
734    mode: Mode,
735    outcome: &kglite::api::session::ExecuteOutcome,
736) -> serde_json::Value {
737    if mode == Mode::Json {
738        serde_json::json!({
739            "ok": true,
740            "rows": exec::outcome_rows_json(outcome),
741        })
742    } else {
743        serde_json::json!({
744            "ok": true,
745            "output": exec::render_outcome(mode, outcome),
746        })
747    }
748}
749
750fn insert_request_id(value: &mut serde_json::Value, request_id: Option<serde_json::Value>) {
751    let Some(id) = request_id else {
752        return;
753    };
754    if let Some(obj) = value.as_object_mut() {
755        obj.entry("id").or_insert(id);
756    }
757}
758
759fn json_error(op: &str, message: String) -> serde_json::Value {
760    serde_json::json!({"ok": false, "op": op, "error": message})
761}
762
763fn request_string(request: &serde_json::Value, key: &str) -> Result<String> {
764    request
765        .get(key)
766        .and_then(|v| v.as_str())
767        .map(str::to_string)
768        .ok_or_else(|| anyhow::anyhow!("missing string field {key:?}"))
769}
770
771fn mode_from_request(request: &serde_json::Value, default_mode: Mode) -> Mode {
772    request
773        .get("format")
774        .and_then(|v| v.as_str())
775        .and_then(Mode::parse)
776        .unwrap_or(default_mode)
777}
778
779fn describe_options_from_json(request: &serde_json::Value) -> Result<DescribeOptions> {
780    Ok(DescribeOptions {
781        types: request.get("types").and_then(json_string_vec),
782        type_search: request
783            .get("type_search")
784            .and_then(|v| v.as_str().map(str::to_string)),
785        connections: detail_from_json(request.get("connections"), detail_connections(false, None))?,
786        cypher: detail_from_json(request.get("cypher"), detail_cypher(false, None))?,
787        fluent: detail_from_json(request.get("fluent"), detail_fluent(false, None))?,
788        max_pairs: request
789            .get("max_pairs")
790            .and_then(|v| v.as_u64())
791            .map(|n| n as usize),
792        sample_truncate: request
793            .get("sample_truncate")
794            .and_then(|v| v.as_u64())
795            .map(|n| n as usize)
796            .or(Some(40)),
797    })
798}
799
800fn json_string_vec(value: &serde_json::Value) -> Option<Vec<String>> {
801    if let Some(s) = value.as_str() {
802        return parse_csv(Some(s));
803    }
804    value.as_array().map(|items| {
805        items
806            .iter()
807            .filter_map(|v| v.as_str().map(str::to_string))
808            .collect()
809    })
810}
811
812trait DetailFromTopics: Sized {
813    fn off() -> Self;
814    fn overview() -> Self;
815    fn topics(topics: Vec<String>) -> Self;
816}
817
818impl DetailFromTopics for ConnectionDetail {
819    fn off() -> Self {
820        ConnectionDetail::Off
821    }
822    fn overview() -> Self {
823        ConnectionDetail::Overview
824    }
825    fn topics(topics: Vec<String>) -> Self {
826        ConnectionDetail::Topics(topics)
827    }
828}
829
830impl DetailFromTopics for CypherDetail {
831    fn off() -> Self {
832        CypherDetail::Off
833    }
834    fn overview() -> Self {
835        CypherDetail::Overview
836    }
837    fn topics(topics: Vec<String>) -> Self {
838        CypherDetail::Topics(topics)
839    }
840}
841
842impl DetailFromTopics for FluentDetail {
843    fn off() -> Self {
844        FluentDetail::Off
845    }
846    fn overview() -> Self {
847        FluentDetail::Overview
848    }
849    fn topics(topics: Vec<String>) -> Self {
850        FluentDetail::Topics(topics)
851    }
852}
853
854fn detail_from_json<T: DetailFromTopics>(
855    value: Option<&serde_json::Value>,
856    default: T,
857) -> Result<T> {
858    match value {
859        None | Some(serde_json::Value::Null) => Ok(default),
860        Some(serde_json::Value::Bool(false)) => Ok(T::off()),
861        Some(serde_json::Value::Bool(true)) => Ok(T::overview()),
862        Some(serde_json::Value::Object(obj)) => detail_from_object(obj),
863        Some(v) => json_string_vec(v)
864            .map(T::topics)
865            .ok_or_else(|| anyhow::anyhow!("detail must be bool, string, string array, or object")),
866    }
867}
868
869fn detail_from_object<T: DetailFromTopics>(
870    obj: &serde_json::Map<String, serde_json::Value>,
871) -> Result<T> {
872    if let Some(types) = obj
873        .get("types")
874        .or_else(|| obj.get("topics"))
875        .or_else(|| obj.get("names"))
876    {
877        return json_string_vec(types)
878            .map(T::topics)
879            .ok_or_else(|| anyhow::anyhow!("detail topics must be string or string array"));
880    }
881
882    let detail = obj
883        .get("detail")
884        .or_else(|| obj.get("mode"))
885        .and_then(|v| v.as_str())
886        .unwrap_or("overview");
887    match detail {
888        "off" | "none" | "false" => Ok(T::off()),
889        "overview" | "true" => Ok(T::overview()),
890        "topics" | "types" => obj
891            .get("value")
892            .or_else(|| obj.get("values"))
893            .and_then(json_string_vec)
894            .map(T::topics)
895            .ok_or_else(|| {
896                anyhow::anyhow!("detail='{detail}' requires value as string or string array")
897            }),
898        other => Err(anyhow::anyhow!(
899            "unknown detail {other:?}; use off, overview, or topics"
900        )),
901    }
902}
903
904fn parse_csv(raw: Option<&str>) -> Option<Vec<String>> {
905    raw.map(|s| {
906        s.split(',')
907            .map(str::trim)
908            .filter(|part| !part.is_empty())
909            .map(str::to_string)
910            .collect()
911    })
912    .filter(|v: &Vec<String>| !v.is_empty())
913}
914
915fn detail_connections(overview: bool, topics: Option<&str>) -> ConnectionDetail {
916    match parse_csv(topics) {
917        Some(v) => ConnectionDetail::Topics(v),
918        None if overview => ConnectionDetail::Overview,
919        None => ConnectionDetail::Off,
920    }
921}
922
923fn detail_cypher(overview: bool, topics: Option<&str>) -> CypherDetail {
924    match parse_csv(topics) {
925        Some(v) => CypherDetail::Topics(v),
926        None if overview => CypherDetail::Overview,
927        None => CypherDetail::Off,
928    }
929}
930
931fn detail_fluent(overview: bool, topics: Option<&str>) -> FluentDetail {
932    match parse_csv(topics) {
933        Some(v) => FluentDetail::Topics(v),
934        None if overview => FluentDetail::Overview,
935        None => FluentDetail::Off,
936    }
937}
938
939fn cypher_string(s: &str) -> String {
940    s.replace('\\', "\\\\").replace('\'', "\\'")
941}
942
943#[cfg(test)]
944mod tests {
945    use super::save_loaded_graph;
946    use kglite::api::io::GraphFileIdentity;
947    use kglite::api::DirGraph;
948    use std::fs;
949    use std::sync::Arc;
950
951    #[test]
952    fn ad_hoc_save_rejects_lost_update() {
953        let tmp = tempfile::tempdir().unwrap();
954        let graph = tmp.path().join("demo.kgl");
955        let mut initial = Arc::new(DirGraph::new());
956        kglite::api::io::save_graph(&mut initial, &graph.to_string_lossy()).unwrap();
957        let mut identity = GraphFileIdentity::capture(&graph).unwrap();
958        let mut working = initial.clone();
959
960        fs::write(&graph, b"competing writer").unwrap();
961        let error = save_loaded_graph(&mut working, &graph, &mut identity, false).unwrap_err();
962
963        assert!(error
964            .to_string()
965            .contains("changed since this session loaded"));
966        assert_eq!(fs::read(&graph).unwrap(), b"competing writer");
967    }
968}