cufflink-cli 0.19.0

use crate::config::CliConfig;
use crate::project_config::ProjectConfig;
use crate::workspace_config::WorkspaceConfig;
use comfy_table::{presets::NOTHING, Table, TableComponent};
use futures_util::StreamExt;
use serde::Deserialize;
use serde_json::Value;
use std::io::{IsTerminal, Write};

/// Maximum number of rows the TTY table preview will hold in memory before
/// showing a truncation hint. Overridden by `--all`.
const TTY_PREVIEW_ROWS: usize = 1000;

/// Run a read-only SQL query against a service's tenant database via the
/// platform `/api/services/{id}/query` endpoint.
#[allow(clippy::too_many_arguments)]
pub async fn run(
    query: &str,
    service: Option<&str>,
    tenant_override: Option<&str>,
    format: Option<&str>,
    all: bool,
    limit: Option<u64>,
    no_pager: bool,
    env: Option<&str>,
) -> eyre::Result<()> {
    let mut config = CliConfig::load_with_env(env)?;
    if let Some(t) = tenant_override {
        config.tenant_override = Some(t.to_string());
    }

    let service_name = resolve_service_name(service)?;
    let service_id = config.find_service_id(&service_name).await?;

    let effective_query = match limited(query, limit) {
        Limited::Wrapped(wrapped) => wrapped,
        Limited::AsTyped(why) => {
            if let Some(why) = why {
                eprintln!("--limit does not apply ({why}); the statement is sent as typed");
            }
            query.to_string()
        }
    };
    let client = config.http_client();
    let resp = config
        .auth_request(
            &client,
            reqwest::Method::POST,
            &format!("{}/api/services/{}/query", config.api_url, service_id),
        )
        .json(&serde_json::json!({ "sql": effective_query }))
        .send()
        .await?;

    if !resp.status().is_success() {
        let status = resp.status();
        let body = resp.text().await.unwrap_or_default();
        eyre::bail!("SQL request failed ({}): {}", status, body);
    }

    let format = format.unwrap_or("table");
    let tty = std::io::stdout().is_terminal();

    match (format, tty) {
        ("jsonl", _) => stream_flat(resp, RowFormat::Jsonl).await,
        ("csv", _) => stream_flat(resp, RowFormat::Csv).await,
        ("tsv", _) => stream_flat(resp, RowFormat::Tsv).await,
        ("table", true) => render_table(resp, all, no_pager).await,
        ("table", false) => stream_flat(resp, RowFormat::Jsonl).await,
        (other, _) => eyre::bail!(
            "Unknown --format '{}'. Use one of: table, jsonl, csv, tsv",
            other
        ),
    }
}

fn resolve_service_name(explicit: Option<&str>) -> eyre::Result<String> {
    if let Some(name) = explicit {
        return Ok(name.to_string());
    }
    let project = ProjectConfig::find_and_load()?.ok_or_else(|| {
        eyre::eyre!(
            "--service not given and no Cufflink.toml found in current directory or any parent."
        )
    })?;

    if let Some(name) = project.service.name.clone() {
        return Ok(name);
    }

    // Rust services leave [service].name empty (it's declared in the Rust
    // `service!` macro). Fall back to the workspace TOML: whichever service
    // path resolves to this project directory is the one we're in.
    if let Some(name) = lookup_name_from_workspace(&project.project_dir) {
        return Ok(name);
    }

    eyre::bail!(
        "Current service has no `name` in Cufflink.toml and cufflink-workspace.toml has no matching entry. Pass --service explicitly."
    )
}

fn lookup_name_from_workspace(project_dir: &std::path::Path) -> Option<String> {
    let (ws, root) = WorkspaceConfig::find_and_load().ok()?;
    match_service_by_path(&ws.services, &root, project_dir)
}

fn match_service_by_path(
    services: &[crate::workspace_config::WorkspaceService],
    root: &std::path::Path,
    project_dir: &std::path::Path,
) -> Option<String> {
    let project_canon = project_dir.canonicalize().ok()?;
    services
        .iter()
        .find(|s| {
            root.join(&s.path)
                .canonicalize()
                .is_ok_and(|p| p == project_canon)
        })
        .map(|s| s.name.clone())
}

/// A piece of a statement that is not a comment or a blank.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Kind {
    Word,
    /// A string, a quoted name or a dollar-quoted body.
    Quoted,
    Semicolon,
    Other,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct Token {
    kind: Kind,
    start: usize,
    end: usize,
}

fn is_word(byte: u8) -> bool {
    byte.is_ascii_alphanumeric() || byte == b'_'
}

/// The end of the string that opens at `start`, where `''` is a quote and,
/// in an `E'…'` string, `\\` escapes what follows it.
fn literal_end(bytes: &[u8], start: usize, escapes: bool) -> usize {
    let mut i = start + 1;
    while i < bytes.len() {
        match (bytes[i], bytes.get(i + 1)) {
            (b'\\', Some(_)) if escapes => i += 2,
            (b'\'', Some(b'\'')) => i += 2,
            (b'\'', _) => return i + 1,
            _ => i += 1,
        }
    }
    bytes.len()
}

/// The end of the quoted name that opens at `start`, where `""` is a quote.
fn quoted_name_end(bytes: &[u8], start: usize) -> usize {
    let mut i = start + 1;
    while i < bytes.len() {
        match (bytes[i], bytes.get(i + 1)) {
            (b'"', Some(b'"')) => i += 2,
            (b'"', _) => return i + 1,
            _ => i += 1,
        }
    }
    bytes.len()
}

/// The `$tag$` that opens at `start`, if one does: a parameter such as
/// `$1` is not one.
fn dollar_tag(query: &str, start: usize) -> Option<&str> {
    let rest = &query.as_bytes()[start + 1..];
    let length = rest.iter().take_while(|b| is_word(**b)).count();
    let opens = rest.first().is_none_or(|b| !b.is_ascii_digit()) && rest.get(length) == Some(&b'$');
    opens.then(|| &query[start..start + length + 2])
}

/// The end of the block comment that opens at `start`; they nest.
fn block_comment_end(bytes: &[u8], start: usize) -> usize {
    let (mut i, mut depth) = (start + 2, 1);
    while i < bytes.len() {
        match (bytes[i], bytes.get(i + 1)) {
            (b'*', Some(b'/')) if depth == 1 => return i + 2,
            (b'*', Some(b'/')) => (i, depth) = (i + 2, depth - 1),
            (b'/', Some(b'*')) => (i, depth) = (i + 2, depth + 1),
            _ => i += 1,
        }
    }
    bytes.len()
}

/// The pieces of `query` as Postgres reads them, without its comments and
/// blanks.
fn tokens(query: &str) -> Vec<Token> {
    let bytes = query.as_bytes();
    let mut found: Vec<Token> = Vec::new();
    let mut i = 0;
    while i < bytes.len() {
        let token = |kind, end| Token {
            kind,
            start: i,
            end,
        };
        let next = bytes.get(i + 1).copied();
        let (skipped, token) = match bytes[i] {
            b if b.is_ascii_whitespace() => (i + 1, None),
            b'-' if next == Some(b'-') => (
                bytes[i..]
                    .iter()
                    .position(|b| *b == b'\n')
                    .map_or(bytes.len(), |line| i + line + 1),
                None,
            ),
            b'/' if next == Some(b'*') => (block_comment_end(bytes, i), None),
            b'\'' => {
                let escaped = found.last().is_some_and(|word| {
                    word.end == i && query[word.start..i].eq_ignore_ascii_case("e")
                });
                let end = literal_end(bytes, i, escaped);
                (end, Some(token(Kind::Quoted, end)))
            }
            b'"' => {
                let end = quoted_name_end(bytes, i);
                (end, Some(token(Kind::Quoted, end)))
            }
            b'$' => match dollar_tag(query, i) {
                Some(tag) => {
                    let body = i + tag.len();
                    let end = query[body..]
                        .find(tag)
                        .map_or(bytes.len(), |close| body + close + tag.len());
                    (end, Some(token(Kind::Quoted, end)))
                }
                None => (i + 1, Some(token(Kind::Other, i + 1))),
            },
            b';' => (i + 1, Some(token(Kind::Semicolon, i + 1))),
            b if is_word(b) => {
                let end = i + bytes[i..].iter().take_while(|b| is_word(**b)).count();
                (end, Some(token(Kind::Word, end)))
            }
            _ => (i + 1, Some(token(Kind::Other, i + 1))),
        };
        found.extend(token);
        i = skipped;
    }
    found
}

/// What `query` holds outside its strings, quoted names and comments: its
/// words, upper-cased, the statements it is, and where the last ends
/// (before the comments, blanks and `;` that trail it).
#[derive(Debug, Default, PartialEq, Eq)]
struct Scan {
    words: Vec<String>,
    statements: usize,
    end: usize,
}

fn scan(query: &str) -> Scan {
    let tokens = tokens(query);
    let statements = tokens
        .split(|token| token.kind == Kind::Semicolon)
        .filter(|statement| !statement.is_empty())
        .count();
    Scan {
        words: tokens
            .iter()
            .filter(|token| token.kind == Kind::Word)
            .map(|word| query[word.start..word.end].to_ascii_uppercase())
            .collect(),
        statements,
        end: tokens
            .iter()
            .rev()
            .find(|token| token.kind != Kind::Semicolon)
            .map_or(0, |last| last.end),
    }
}

/// The words of a statement that writes, or makes a table, which a
/// subquery cannot hold.
const WRITES: [&str; 5] = ["INSERT", "UPDATE", "DELETE", "MERGE", "INTO"];

/// What `--limit` makes of what was typed.
#[derive(Debug, PartialEq, Eq)]
enum Limited {
    /// One query that writes nothing, held to the limit by wrapping it.
    Wrapped(String),
    /// Sent as typed, and why the limit does not apply.
    AsTyped(Option<&'static str>),
}

/// `query` held to `limit` rows by wrapping it, so a `LIMIT` of its own, a
/// trailing comment or a trailing `;` changes nothing. Only one statement
/// that answers rows and writes nothing is wrapped: several statements, a
/// statement that is not a query, a data-modifying `WITH` and a
/// `SELECT … INTO` are sent as typed.
fn limited(query: &str, limit: Option<u64>) -> Limited {
    let Some(limit) = limit else {
        return Limited::AsTyped(None);
    };
    let scan = scan(query);
    let answers_rows = scan
        .words
        .first()
        .is_some_and(|first| ["SELECT", "WITH", "VALUES", "TABLE"].contains(&first.as_str()));
    if scan.statements > 1 {
        Limited::AsTyped(Some("it is several statements"))
    } else if !answers_rows {
        Limited::AsTyped(Some("it is not a query"))
    } else if scan
        .words
        .iter()
        .any(|word| WRITES.contains(&word.as_str()))
    {
        Limited::AsTyped(Some("it writes"))
    } else {
        Limited::Wrapped(format!(
            "SELECT * FROM (\n{}\n) q LIMIT {limit}",
            &query[..scan.end]
        ))
    }
}

#[derive(Debug, Deserialize)]
struct ColumnMeta {
    name: String,
}

#[derive(Copy, Clone)]
enum RowFormat {
    Jsonl,
    Csv,
    Tsv,
}

enum Frame {
    Columns(Vec<ColumnMeta>),
    Row(Vec<Value>),
    Done {
        rows: u64,
        truncated: bool,
        elapsed_ms: u64,
    },
    Error(String),
}

fn parse_frame(line: &str) -> eyre::Result<Frame> {
    let v: Value = serde_json::from_str(line)
        .map_err(|e| eyre::eyre!("Malformed NDJSON frame: {} — line was: {}", e, line))?;
    match v.get("type").and_then(|t| t.as_str()) {
        Some("columns") => {
            let cols: Vec<ColumnMeta> =
                serde_json::from_value(v.get("columns").cloned().unwrap_or(Value::Array(vec![])))?;
            Ok(Frame::Columns(cols))
        }
        Some("row") => {
            let values: Vec<Value> = v
                .get("values")
                .and_then(|x| x.as_array())
                .cloned()
                .unwrap_or_default();
            Ok(Frame::Row(values))
        }
        Some("done") => Ok(Frame::Done {
            rows: v.get("rows").and_then(|x| x.as_u64()).unwrap_or(0),
            truncated: v
                .get("truncated")
                .and_then(|x| x.as_bool())
                .unwrap_or(false),
            elapsed_ms: v.get("elapsed_ms").and_then(|x| x.as_u64()).unwrap_or(0),
        }),
        Some("error") => {
            let msg = v
                .get("message")
                .and_then(|x| x.as_str())
                .unwrap_or("(no message)")
                .to_string();
            Ok(Frame::Error(msg))
        }
        other => eyre::bail!("Unknown NDJSON frame type: {:?}", other),
    }
}

/// Stream every row directly to stdout in the chosen row format.
/// No buffering beyond line boundaries; safe to pipe into `head`, `jq`, etc.
async fn stream_flat(resp: reqwest::Response, kind: RowFormat) -> eyre::Result<()> {
    let mut stream = resp.bytes_stream();
    let mut buf: Vec<u8> = Vec::new();
    let stdout = std::io::stdout();
    let mut out = stdout.lock();
    let mut columns: Vec<ColumnMeta> = Vec::new();

    while let Some(chunk) = stream.next().await {
        let chunk = chunk?;
        buf.extend_from_slice(&chunk);
        while let Some(pos) = buf.iter().position(|&b| b == b'\n') {
            let line_bytes: Vec<u8> = buf.drain(..=pos).collect();
            let line = std::str::from_utf8(&line_bytes[..line_bytes.len() - 1])?.trim();
            if line.is_empty() {
                continue;
            }
            match parse_frame(line)? {
                Frame::Columns(c) => {
                    match kind {
                        RowFormat::Csv => writeln!(out, "{}", join_header(&c, ','))?,
                        RowFormat::Tsv => writeln!(out, "{}", join_header(&c, '\t'))?,
                        RowFormat::Jsonl => {}
                    }
                    columns = c;
                }
                Frame::Row(values) => match kind {
                    RowFormat::Jsonl => {
                        let obj = if columns.is_empty() {
                            Value::Array(values)
                        } else {
                            let mut map = serde_json::Map::with_capacity(columns.len());
                            for (c, v) in columns.iter().zip(values.into_iter()) {
                                map.insert(c.name.clone(), v);
                            }
                            Value::Object(map)
                        };
                        writeln!(out, "{}", serde_json::to_string(&obj)?)?;
                    }
                    RowFormat::Csv => writeln!(out, "{}", encode_row(&values, ','))?,
                    RowFormat::Tsv => writeln!(out, "{}", encode_row(&values, '\t'))?,
                },
                Frame::Done {
                    rows, truncated, ..
                } => {
                    if truncated {
                        eprintln!("(server row cap reached after {rows} rows)");
                    }
                }
                Frame::Error(msg) => eyre::bail!("Server error during streaming: {msg}"),
            }
        }
    }
    Ok(())
}

/// Buffer rows (up to TTY_PREVIEW_ROWS unless `--all`) and render a table.
async fn render_table(resp: reqwest::Response, all: bool, no_pager: bool) -> eyre::Result<()> {
    let mut stream = resp.bytes_stream();
    let mut buf: Vec<u8> = Vec::new();
    let mut columns: Vec<ColumnMeta> = Vec::new();
    let mut rows: Vec<Vec<Value>> = Vec::new();
    let mut skipped: u64 = 0;
    let mut total: u64 = 0;
    let mut server_truncated = false;
    let mut elapsed_ms: u64 = 0;

    while let Some(chunk) = stream.next().await {
        let chunk = chunk?;
        buf.extend_from_slice(&chunk);
        while let Some(pos) = buf.iter().position(|&b| b == b'\n') {
            let line_bytes: Vec<u8> = buf.drain(..=pos).collect();
            let line = std::str::from_utf8(&line_bytes[..line_bytes.len() - 1])?.trim();
            if line.is_empty() {
                continue;
            }
            match parse_frame(line)? {
                Frame::Columns(c) => columns = c,
                Frame::Row(values) => {
                    total += 1;
                    if all || rows.len() < TTY_PREVIEW_ROWS {
                        rows.push(values);
                    } else {
                        skipped += 1;
                    }
                }
                Frame::Done {
                    rows: server_rows,
                    truncated,
                    elapsed_ms: elapsed,
                } => {
                    server_truncated = truncated;
                    elapsed_ms = elapsed;
                    if server_rows > 0 && server_rows != total {
                        // Trust the server count over our row accumulator.
                        total = server_rows;
                    }
                }
                Frame::Error(msg) => eyre::bail!("Server error during streaming: {msg}"),
            }
        }
    }

    if columns.is_empty() && rows.is_empty() {
        println!("(empty result)");
        return Ok(());
    }

    let mut table = Table::new();
    table.load_preset(NOTHING);
    table.set_style(TableComponent::HeaderLines, '-');
    table.set_style(TableComponent::MiddleHeaderIntersections, ' ');
    let header: Vec<String> = columns.iter().map(|c| c.name.clone()).collect();
    table.set_header(header);
    for row in &rows {
        let cells: Vec<String> = row.iter().map(cell_string).collect();
        table.add_row(cells);
    }

    let rendered = format!("{table}");

    if all && !no_pager && !rendered.is_empty() {
        pipe_to_pager(&rendered).unwrap_or_else(|_| println!("{rendered}"));
    } else {
        println!("{rendered}");
    }

    let hint = match (skipped > 0, server_truncated) {
        (true, true) => Some(format!(
            "{skipped} rows hidden. Server cap hit at {total}. Use --format csv|jsonl to stream, or reduce the query."
        )),
        (true, false) => Some(format!(
            "{skipped} rows hidden (showing first {TTY_PREVIEW_ROWS}). Use --all, --limit N, or --format csv|jsonl."
        )),
        (false, true) => Some(format!(
            "Server cap hit at {total} rows. Query returned more than the platform allows."
        )),
        (false, false) => None,
    };
    if let Some(h) = hint {
        eprintln!("{h}");
    }
    if elapsed_ms > 0 {
        eprintln!("({total} rows, {elapsed_ms} ms)");
    }

    Ok(())
}

fn cell_string(v: &Value) -> String {
    match v {
        Value::Null => String::new(),
        Value::String(s) => s.clone(),
        other => other.to_string(),
    }
}

fn join_header(cols: &[ColumnMeta], sep: char) -> String {
    cols.iter()
        .map(|c| csv_escape(&c.name, sep))
        .collect::<Vec<_>>()
        .join(&sep.to_string())
}

fn encode_row(values: &[Value], sep: char) -> String {
    values
        .iter()
        .map(|v| csv_escape(&cell_string(v), sep))
        .collect::<Vec<_>>()
        .join(&sep.to_string())
}

/// Minimal RFC 4180-style escaping for CSV/TSV cells.
fn csv_escape(s: &str, sep: char) -> String {
    if s.contains(sep) || s.contains('"') || s.contains('\n') || s.contains('\r') {
        let escaped = s.replace('"', "\"\"");
        format!("\"{escaped}\"")
    } else {
        s.to_string()
    }
}

fn pipe_to_pager(content: &str) -> eyre::Result<()> {
    let pager = std::env::var("PAGER").unwrap_or_else(|_| "less -SRX".to_string());
    let mut parts = pager.split_whitespace();
    let program = parts
        .next()
        .ok_or_else(|| eyre::eyre!("PAGER env var is empty"))?;
    let args: Vec<&str> = parts.collect();

    let mut child = std::process::Command::new(program)
        .args(&args)
        .stdin(std::process::Stdio::piped())
        .spawn()?;
    if let Some(mut stdin) = child.stdin.take() {
        stdin.write_all(content.as_bytes())?;
    }
    child.wait()?;
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn a_limit_wraps_the_query() {
        for (query, limited_to) in [
            (
                "select * from items",
                "SELECT * FROM (\nselect * from items\n) q LIMIT 10",
            ),
            (
                "select * from items; \n",
                "SELECT * FROM (\nselect * from items\n) q LIMIT 10",
            ),
            (
                "SELECT * FROM items LIMIT 500",
                "SELECT * FROM (\nSELECT * FROM items LIMIT 500\n) q LIMIT 10",
            ),
            (
                "select * from items -- all of them",
                "SELECT * FROM (\nselect * from items\n) q LIMIT 10",
            ),
            (
                "select 1; -- note",
                "SELECT * FROM (\nselect 1\n) q LIMIT 10",
            ),
            (
                "select 1 /* one */ ; ; /* done */\n",
                "SELECT * FROM (\nselect 1\n) q LIMIT 10",
            ),
            (
                "select name from limits where note = ' LIMIT '",
                "SELECT * FROM (\nselect name from limits where note = ' LIMIT '\n) q LIMIT 10",
            ),
            (
                "select 'it''s; -- not a comment' as a",
                "SELECT * FROM (\nselect 'it''s; -- not a comment' as a\n) q LIMIT 10",
            ),
            (
                "select \"delete\", 'insert into x' from \"update\"",
                "SELECT * FROM (\nselect \"delete\", 'insert into x' from \"update\"\n) q LIMIT 10",
            ),
            (
                "select $$ ; delete $$, $tag$ it's $tag$",
                "SELECT * FROM (\nselect $$ ; delete $$, $tag$ it's $tag$\n) q LIMIT 10",
            ),
            (
                "select E'\\'; delete'",
                "SELECT * FROM (\nselect E'\\'; delete'\n) q LIMIT 10",
            ),
            (
                "-- the newest\n/* first */ WITH n AS (SELECT 1) SELECT * FROM n",
                "SELECT * FROM (\n-- the newest\n/* first */ WITH n AS (SELECT 1) SELECT * FROM n\n) q LIMIT 10",
            ),
            (
                "values (1), (2)",
                "SELECT * FROM (\nvalues (1), (2)\n) q LIMIT 10",
            ),
            (
                "TABLE items",
                "SELECT * FROM (\nTABLE items\n) q LIMIT 10",
            ),
            (
                "( select 1) union (select 2)",
                "SELECT * FROM (\n( select 1) union (select 2)\n) q LIMIT 10",
            ),
            (
                "select * from t where id = $1",
                "SELECT * FROM (\nselect * from t where id = $1\n) q LIMIT 10",
            ),
        ] {
            assert_eq!(
                limited(query, Some(10)),
                Limited::Wrapped(limited_to.to_string()),
                "{query}"
            );
        }
    }

    #[test]
    fn only_one_query_that_writes_nothing_is_limited() {
        for (statement, why) in [
            ("show search_path", "it is not a query"),
            ("explain select * from items", "it is not a query"),
            ("selector", "it is not a query"),
            ("-- select\nshow search_path", "it is not a query"),
            ("/* unterminated select", "it is not a query"),
            ("/* nested /* select */ */ show all", "it is not a query"),
            ("", "it is not a query"),
            (
                "WITH d AS (DELETE FROM items RETURNING *) SELECT * FROM d",
                "it writes",
            ),
            (
                "with moved as (update items set n = 1 returning id) select * from moved",
                "it writes",
            ),
            ("SELECT * INTO copy FROM items", "it writes"),
            ("select 1; select 2", "it is several statements"),
            ("select 1; -- one\nselect 2;", "it is several statements"),
        ] {
            assert_eq!(
                limited(statement, Some(10)),
                Limited::AsTyped(Some(why)),
                "{statement}"
            );
        }
        assert_eq!(limited("select 1;", None), Limited::AsTyped(None));
    }

    #[test]
    fn parses_columns_frame() {
        let line = r#"{"type":"columns","columns":[{"name":"id","type_name":"uuid"},{"name":"email","type_name":"text"}]}"#;
        match parse_frame(line).unwrap() {
            Frame::Columns(c) => {
                assert_eq!(c.len(), 2);
                assert_eq!(c[0].name, "id");
                assert_eq!(c[1].name, "email");
            }
            _ => panic!("expected Columns"),
        }
    }

    #[test]
    fn parses_row_frame() {
        let line = r#"{"type":"row","values":["abc",42,null,true]}"#;
        match parse_frame(line).unwrap() {
            Frame::Row(vs) => {
                assert_eq!(vs.len(), 4);
                assert_eq!(vs[1].as_i64().unwrap(), 42);
                assert!(vs[2].is_null());
            }
            _ => panic!("expected Row"),
        }
    }

    #[test]
    fn parses_done_frame() {
        let line = r#"{"type":"done","rows":17,"truncated":true,"elapsed_ms":250}"#;
        match parse_frame(line).unwrap() {
            Frame::Done {
                rows,
                truncated,
                elapsed_ms,
            } => {
                assert_eq!(rows, 17);
                assert!(truncated);
                assert_eq!(elapsed_ms, 250);
            }
            _ => panic!("expected Done"),
        }
    }

    #[test]
    fn csv_escape_quotes_embedded_commas() {
        assert_eq!(csv_escape("hello", ','), "hello");
        assert_eq!(csv_escape("a,b", ','), "\"a,b\"");
        assert_eq!(csv_escape("a\"b", ','), "\"a\"\"b\"");
        assert_eq!(csv_escape("multi\nline", ','), "\"multi\nline\"");
        // TSV doesn't escape commas.
        assert_eq!(csv_escape("a,b", '\t'), "a,b");
        assert_eq!(csv_escape("a\tb", '\t'), "\"a\tb\"");
    }

    #[test]
    fn cell_string_flattens_null_to_empty() {
        assert_eq!(cell_string(&Value::Null), "");
        assert_eq!(cell_string(&Value::String("x".into())), "x");
        assert_eq!(cell_string(&Value::Bool(true)), "true");
    }

    #[test]
    fn match_service_by_path_resolves_rust_service() {
        use crate::workspace_config::WorkspaceService;

        let tmp = tempfile::tempdir().unwrap();
        let root = tmp.path();
        std::fs::create_dir_all(root.join("backend/asset-service")).unwrap();
        std::fs::create_dir_all(root.join("apps/portal")).unwrap();

        let services = vec![
            WorkspaceService {
                name: "asset-service".into(),
                path: "backend/asset-service".into(),
                seed: None,
                preview_tables: None,
            },
            WorkspaceService {
                name: "portal".into(),
                path: "apps/portal".into(),
                seed: None,
                preview_tables: None,
            },
        ];

        let got = match_service_by_path(&services, root, &root.join("backend/asset-service"));
        assert_eq!(got.as_deref(), Some("asset-service"));

        let got = match_service_by_path(&services, root, &root.join("apps/portal"));
        assert_eq!(got.as_deref(), Some("portal"));

        let got = match_service_by_path(&services, root, &root.join("does/not/exist"));
        assert!(got.is_none());
    }
}